from __future__ import annotations
import copy
import datetime
import functools
import json
import logging
import math
import os
import traceback
from collections import defaultdict, namedtuple, deque
from typing import Callable, DefaultDict, Dict, List, Optional, Tuple, Union

import google.protobuf.json_format as json_format
import numpy as np
import pandas as pd
import pytz
import scipy.stats

import coin.proto.coin_order_enums_pb2 as order_enums
import coin.proto.coin_strategy_reporter_pb2 as strategy_reporter_proto
import coin.proto.options_pricing_pb2 as op_pb2
import coin.support.options_pricing.bs as bs_model
from coin.exchange.base.kr_rest.py_product import PyProduct
from coin.base.datetime_util import to_datetime, to_timestamp_int
from coin.base.git_util import fetch_git_commit_sha_datetime
from coin.base.timestamp import get_timestamp
from coin.exchange.base.proto_logger import ProtoLogProvider
from coin.exchange.kr_rest.product.product_impl import generate_product_from_str2
from coin.proto.coin_feed_pb2 import BookEntry
from coin.proto.coin_order_gateway_pb2 import OrderGatewayLog, OrderEvent
from coin.proto.coin_market_query_pb2 import (
    ExchangeTicker, ProductKline, ProductTicker)
from coin.proto.coin_query_pb2 import (AccountBalance,
                                       AccountPosition,
                                       CurrencyBalance,
                                       ProductPosition)
from coin.proto.coin_strategy_pb2 import StrategyLog
from coin.proto.coin_strategy_summary_pb2 import StrategyLogInfo
from coin.proto.coin_telemetry_pb2 import StrategyLatencyStatsProto
from coin.proto.coin_strategy_summary_pb2 import (AccountStrategyStat,
                                                  StrategySummary,
                                                  StrategyDetailEntry,
                                                  StrategyDetailSummary,
                                                  StrategySummarySourceInfoProto,
                                                  StrategyStat,
                                                  StrategyStatLog)
from coin.proto.coin_executor_pb2 import (AggressiveExecutorConfig, PassiveExecutorConfig)
from coin.proto.coin_request_pb2 import (AccountInfoProto,
                                         AccountRequestProto,
                                         CalculatorInfoProto,
                                         ExchangeApiInfoProto,
                                         ProcessInfoProto,
                                         StrategyRequestProto,
                                         StrategyInfoProto)
from coin.strategy.accounting.fee import get_fee_rate
from coin.support.accounting.logic.position_estimate import calculate_position_value
from coin.support.accounting.logic.query_util import AccountRequest
from coin.support.options_pricing.types import OptionsGreeks, PricingParamsType
from coin.support.options_pricing.util.pricing_util import calculate_sabr_greeks
from coin.support.pta.logging.strategy_stat_logger import StrategyStatProvider
from coin.support.pta.logic.pta_stats_calculator import (
    calculate_turnover,
    calculate_turnover_in_usd,
    get_accounting_currency,
    get_order_sign,
    is_inverse_product,
    get_product_multiplier)
from coin.support.telemetry.app.latency_analyser import LatencyStatsCalculator
from coin.tool.strat_monitor.plot.strat_plotter import (
    StrategyOrderPlotMsg,
    SymbolOrderPlotMsg)
from xunkemgmt_client.support.accounting.database.db_output import DBOutputKline


NS_4HR = 4 * 3600 * 10**9
NS_8HR = 8 * 3600 * 10**9
TICKER_EXPIRE_TIME = 24 * 60 * 60 * 10**9
AccountKeyType = Tuple[str, str, str]
ExecutorType = Union[AggressiveExecutorConfig, PassiveExecutorConfig]


def _isclose(a, b, abs_tol=None):
  abs_tol = abs_tol if abs_tol is not None else 10**-9
  return math.isclose(a, b, abs_tol=abs_tol)


def _adjust_by_pnl_balance_diff(strategy_name, trading_date):
  ret = False
  if trading_date == datetime.date(2021, 9, 7):
    ret = True
  return ret

def _adjust_by_reserve_diff(strategy_name, trading_date):
  ret = False
  # special treatment due to Binance spot balance glitch
  if datetime.date(2021, 8, 13) <= trading_date <= datetime.date(2021, 8, 17):
    strat_list = [
        'lm_agg2_binance_spot_btcquote_1',
        'lm_agg2_binance_spot_btcquote_2',
        'lm_agg2_binance_spot_busd',
        'lm_agg2_binance_spot_eur',
        'lm_agg2_binance_spot_gbp',
        'lm_agg2_binance_spot_usdc',
        'lm_agg2_binance_spot_eth',
        'lm_agg2_binance_spot_minor_1',
        'lm_agg2_binance_spot_minor_2',
        'lm_agg2_binance_spot_minor_3',
        'lm_agg2_binance_spot_minor_4',
        'lm_agg2_binance_spot_minor_5',
        'lm_agg2_binance_spot_minor_6',
        'lm_agg2_binance_spot_minor_7',
        'lm_agg2_binance_spot_minor_8',
        'lm_focus_binance_spot_2',
        'lm_focus_binance_spot_1',
        'lm_volume_binance_busd_spot_1',
        'lm_volume_binance_eur_spot_1',
        'lm_volume_binance_eur_spot_2',
        'lm_volume_binance_usdt_spot_1'
    ]
    if strategy_name in strat_list:
      ret = True
  if strategy_name.startswith('vmm_') or strategy_name.startswith('hamm_'):
    ret = True
  return ret


def _multi_window_pnl_to_strategy_stat(pnl_stats, pnl_type):
  strat_stats = []
  assert isinstance(pnl_stats, MultiWindowStats)
  pnl_mark = pnl_stats.stats_mark
  init = pnl_mark.init
  for (ts, pnl_balance) in pnl_mark.history:
    if init is not None:
      if pnl_type == 'from_balance':
        pnl_stat = StrategyStat(stat_ts=ts, pnl_mark=pnl_balance - init,
                                pnl_mark_from_balance=pnl_balance - init)
      elif pnl_type == 'from_fill':
        pnl_stat = StrategyStat(stat_ts=ts, pnl_mark_from_fill=pnl_balance - init)
      else:
        raise ValueError('Unknown pnl_type: %s' % pnl_type)
      strat_stats.append(pnl_stat)
  return strat_stats


def _get_account_key(account_request) -> AccountKeyType:
  assert isinstance(account_request, AccountRequestProto)
  assert account_request.HasField('market_type') and \
         account_request.HasField('exchange') and \
         account_request.HasField('owner'), account_request
  return (account_request.market_type,
          account_request.exchange,
          account_request.owner)


def _get_reserve_dict(reserve: AccountBalance):
  assert isinstance(reserve, AccountBalance)
  reserve_dict = {}
  for balance in reserve.each_balance:
    if balance.HasField('total') and balance.HasField('currency'):
      reserve_dict[balance.currency] = balance.total
  if not reserve_dict:
    logging.info('Reserve is empty. %s' % reserve)
  return reserve_dict


def _extract_account_request(pb):
  account_request = None
  if pb.HasField('market_type') and \
     pb.HasField('exchange') and \
     pb.HasField('owner'):
    account_request = AccountRequestProto(
        market_type=pb.market_type,
        exchange=pb.exchange,
        owner=pb.owner)
  return account_request


def get_fill_type_inferred(order_event):
  assert order_event.type == OrderEvent.ORDER_FILLED
  fill_type = order_enums.UNKNOWN_FILL_TYPE
  if order_event.HasField('guess_fill_type'):
    fill_type = order_event.guess_fill_type
  if fill_type == order_enums.UNKNOWN_FILL_TYPE:
    order_price = \
        order_event.order_price if order_event.HasField('order_price') else None
    fill_price = order_event.fill_price
    if order_price is not None and fill_price is not None:
      if _isclose(order_price, fill_price):
        fill_type = order_enums.MAKER_FILL_TYPE
      else:
        fill_type = order_enums.TAKER_FILL_TYPE
  return fill_type


def get_fill_type(order_event):
  assert order_event.type == OrderEvent.ORDER_FILLED
  fill_type = order_enums.UNKNOWN_FILL_TYPE
  if order_event.HasField('fill_type'):
    fill_type = order_event.fill_type
  return fill_type


def _is_maker(order_event):
  is_maker = False
  assert order_event.type == OrderEvent.ORDER_FILLED
  fill_type = get_fill_type(order_event)
  if fill_type == order_enums.UNKNOWN_FILL_TYPE:
    fill_type = get_fill_type_inferred(order_event)
  if fill_type == order_enums.MAKER_FILL_TYPE:
    is_maker = True
  return is_maker


def _is_expired(product, current_ts):
  expired = False
  if product.market_name in ('Futures', 'Options'):
    if (not product.is_perpetual and not product.is_immediate):
      current_date = to_datetime(current_ts).date()
      expiry_date = datetime.datetime.strptime(product.expiry_str, '%Y%m%d').date()
      if current_date > expiry_date:
        expired = True
      elif current_date == expiry_date:
        try:
          generate_product_from_str2(
              product.market_name,
              product.exchange_name,
              None,
              product.symbol,
              to_datetime(current_ts))
        except Exception:
          expired = True
  return expired


def is_executor_config_updated(lfs: ExecutorType, rfs: ExecutorType):
  updated = False
  if lfs.min_pos != rfs.min_pos or \
     lfs.max_pos != rfs.max_pos or \
     lfs.reserve != rfs.reserve:
    updated = True
  return updated


def _get_accounting_currency(
    market_type: str, exchange: str, symbol: str, ts: int) -> str:
  dt = datetime.datetime.utcfromtimestamp(ts / 10**9)
  return get_accounting_currency(market_type, exchange, symbol, dt)


def _gen_product(market_type, exchange, symbol, ts):
  dt = datetime.datetime.utcfromtimestamp(ts / 10**9)
  return generate_product_from_str2(market_type, exchange, None, symbol, dt)


def _get_today_timestamp(ts):
  dt = datetime.datetime.utcfromtimestamp(ts / 10**9).date()
  today_dt = to_datetime(dt).replace(tzinfo=pytz.UTC)
  today_ts = int(today_dt.timestamp() * 10**9)
  return today_ts


def _create_multi_window_stats(*args, **kwargs):
  return MultiWindowStats(*args, **kwargs)


def _check_start_ts_and_window(window, start_ts):
  if start_ts is None:
    assert window is not None
  else:
    assert window is None


def _gen_latency_stats_proto(process, strategy, latency_stats):
  stats_proto = StrategyLatencyStatsProto(process=process, strategy=strategy)
  stats_proto.MergeFrom(latency_stats)
  return stats_proto


def _calculate_otc_options_price(product: PyProduct,
                                 position: ProductPosition,
                                 exchange_ticker: ExchangeTicker,
                                 current_timestamp: int):
    assert exchange_ticker.HasField('exchange')
    underlying_product = position.underlying_product
    underlying_exchange = None
    underlying_symbol = '%s-%s' % (
        product.base.currency, product.quote.currency)
    if underlying_product.HasField('symbol'):
      underlying_symbol = underlying_product.symbol
    if underlying_product.HasField('exchange'):
      underlying_exchange = underlying_product.exchange
    last_prices = []
    if underlying_exchange is None or \
       underlying_exchange == exchange_ticker.exchange:
      for each_ticker in exchange_ticker.each_ticker:
        if each_ticker.symbol == underlying_symbol:
          last_prices.append(each_ticker.last)
    strike_price = product.strike_price
    volatility = position.volatility if position.HasField('volatility') else None
    rfr = position.risk_free_rate if position.HasField('risk_free_rate') else None
    expiry = datetime.datetime.strptime(product.expiry_str, '%Y%m%d')
    options_value = None
    if len(last_prices) > 0 and volatility is not None and rfr is not None:
      options_value = bs_model.calculate_bs_price(
          underlying_price=last_prices[0],
          risk_free_rate=rfr,
          implied_volatility=volatility,
          strike_price=strike_price,
          maturity=to_timestamp_int(expiry),
          side=product.option_side,
          current_timestamp=current_timestamp)
    return options_value


class MultiWindowStats(object):
  def __init__(self, current_timestamp, *,
               start_ts=None, update_sum=False, store_data=False,
               stats_mark=False, stats_4hr=False, stats_8hr=False):
    self._stats_mark = TimeWindowStats(current_timestamp,
                                       start_ts=start_ts,
                                       update_sum=update_sum,
                                       store_data=store_data) if stats_mark else None
    self._stats_4hr = TimeWindowStats(current_timestamp,
                                      window=NS_4HR,
                                      update_sum=update_sum,
                                      store_data=store_data) if stats_4hr else None
    self._stats_8hr = TimeWindowStats(current_timestamp,
                                      window=NS_8HR,
                                      update_sum=update_sum,
                                      store_data=store_data) if stats_8hr else None

  @property
  def stats_mark(self):
    return self._stats_mark

  @property
  def stats_4hr(self):
    return self._stats_4hr

  @property
  def stats_8hr(self):
    return self._stats_8hr

  def update(self, ts, elem=None, start_ts=None):
    for stats in [self._stats_mark, self._stats_4hr, self._stats_8hr]:
      if stats is not None:
        stats.update(ts, elem=elem, start_ts=start_ts)

  def register_stats_mark_observer(self, callback: Callable[[int, float], None]):
    self._stats_mark.register_stats_mark_observer(callback)


class TimeWindowStats(object):
  def __init__(self, current_timestamp, *,
               window=None, start_ts=None, update_sum=False, store_data=False):
    _check_start_ts_and_window(window, start_ts)
    self._current_ts = current_timestamp
    self._window = window
    self._start_ts = start_ts
    self._update_sum = update_sum
    self._store_data = store_data
    self._sum = 0.
    self._deque = deque()
    self._last_popleft_elem = None
    self._start_ts_changed = False
    self._callbacks: List[Callable[[int, float], None]] = []

  def __len__(self):
    return len(self._deque)

  def _update_ts(self, ts, start_ts):
    if ts > self._current_ts:
      self._current_ts = ts
    if (start_ts is not None and
        self._start_ts is not None and
        self._start_ts != start_ts):
      self._start_ts_changed = True
      self._start_ts = start_ts
    else:
      self._start_ts_changed = False

  @property
  def init(self):
    if (self._start_ts is not None and self.last_popleft is not None):
      init = self.last_popleft
    else:
      init = self.first
    return init

  @property
  def last_popleft(self):
    last_popleft = None
    if self._last_popleft_elem is not None:
      _, last_popleft = self._last_popleft_elem
    return last_popleft

  @property
  def first(self):
    first = None
    if len(self._deque) > 0:
      _, first = self._deque[0]
    return first

  @property
  def last(self):
    last = None
    if len(self._deque) > 0:
      _, last = self._deque[-1]
    else:
      last = self.last_popleft
    return last

  @property
  def first_elem(self):
    first_elem = None
    if len(self._deque) > 0:
      first_elem = self._deque[0]
    return first_elem

  @property
  def last_elem(self):
    last_elem = None
    if len(self._deque) > 0:
      last_elem = self._deque[-1]
    else:
      last_elem = self._last_popleft_elem
    return last_elem

  @property
  def sum(self):
    return self._sum

  @property
  def diff(self):
    if len(self._deque) > 0:
      first = self.init
      last = self.last
      diff = last - first
    else:
      diff = 0
    return diff

  @property
  def average(self):
    self._check_store_data()
    average = None
    if len(self._deque) > 0:
      average = self._sum / len(self._deque)
    return average

  @property
  def history(self):
    self._check_store_data()
    return list(self._deque)

  def pop(self):
    return self._deque.pop()

  def register_stats_mark_observer(self, callback: Callable[[int, float], None]):
    self._callbacks.append(callback)

  def update(self, ts, elem=None, start_ts=None):
    self._update_ts(ts, start_ts)
    size = len(self._deque)
    if elem is not None:
      self._deque.append((ts, elem))
      if self._update_sum:
        self._sum += elem
    window = self._current_ts - self._start_ts if self._window is None else self._window
    while self._deque:
      elem_ts, elem_value = self._deque[0]
      if self._current_ts - elem_ts <= window:
        break
      elem = self._deque.popleft()
      self._last_popleft_elem = elem
      if self._update_sum:
        self._sum -= elem_value

    publish_data = []
    if self._start_ts is not None:
      if self._start_ts_changed:
        publish_data = self._deque
      elif len(self._deque) > size:
        size_diff = size - len(self._deque)
        assert size_diff == -1, size_diff
        publish_data = [self._deque[size_diff]]
      for elem in publish_data:
        for callback in self._callbacks:
          callback(*elem)
    self._clean_data()

  def _check_store_data(self):
    if self._start_ts is not None:
      assert self._store_data

  def _clean_data(self):
    if self._start_ts is not None and not self._store_data:
      # only keep the first and last elem
      if len(self._deque) > 1:
        self._deque = deque([self._deque[0], self._deque[-1]])


class FullPartialFrCalculator(object):
  def __init__(self, current_ts, start_ts):
    self._pfill_qty = _create_multi_window_stats(
        current_ts, start_ts=start_ts, update_sum=True, stats_mark=True, stats_8hr=True)
    self._ffill_qty = _create_multi_window_stats(
        current_ts, start_ts=start_ts, update_sum=True, stats_mark=True, stats_8hr=True)
    self._sub_qty = _create_multi_window_stats(
        current_ts, start_ts=start_ts, update_sum=True, stats_mark=True, stats_8hr=True)
    self._subfill_qty_map = defaultdict(lambda: [None, 0])
    self._dead_order_time = {}

  def update_by_order_event(self, ts, order_event):
    assert isinstance(order_event, OrderEvent)
    if not order_event.HasField('external_order_id'):
      return
    sub_fill_info = self._subfill_qty_map[order_event.external_order_id]

    if order_event.order_state == order_enums.DEAD_ORDER:
      if sub_fill_info[0] is None:
        assert order_event.HasField('order_qty')
        sub_fill_info[0] = order_event.order_qty
    if order_event.type == OrderEvent.ORDER_FILLED:
      sub_fill_info[1] += order_event.fill_qty

    # first death
    if (order_event.order_state == order_enums.DEAD_ORDER
        and order_event.external_order_id not in self._dead_order_time):
      self._dead_order_time[order_event.external_order_id] = ts
      if self._subfill_qty_map[order_event.external_order_id][0] is not None:
        if _isclose(sub_fill_info[0], sub_fill_info[1]):
          self._ffill_qty.update(ts, sub_fill_info[1])
        else:
          self._pfill_qty.update(ts, sub_fill_info[1])
        self._sub_qty.update(ts, sub_fill_info[0])

  def get_stats(self):
    ffqty_mark = self._ffill_qty.stats_mark.sum
    pfqty_mark = self._pfill_qty.stats_mark.sum
    subqty_mark = self._sub_qty.stats_mark.sum
    ffqty_8hr = self._ffill_qty.stats_8hr.sum
    pfqty_8hr = self._pfill_qty.stats_8hr.sum
    subqty_8hr = self._sub_qty.stats_8hr.sum
    ffqtyrat_8hr = None if subqty_8hr == 0 else ffqty_8hr / subqty_8hr
    pfqtyrat_8hr = None if subqty_8hr == 0 else pfqty_8hr / subqty_8hr
    ffqtyrat_mark = None if subqty_mark == 0 else ffqty_mark / subqty_mark
    pfqtyrat_mark = None if subqty_mark == 0 else pfqty_mark / subqty_mark
    return ffqtyrat_mark, pfqtyrat_mark, ffqtyrat_8hr, pfqtyrat_8hr


class StrategyStatLogPublisher(object):
  def __init__(self, *,
               account_request: Optional[AccountRequestProto]=None,
               symbol: Optional[str]=None,
               ts: int=None,):
    self._account_request = account_request
    self._symbol = symbol
    self._accounting_currency = None
    if symbol is not None:
      self._accounting_currency = _get_accounting_currency(
          self._account_request.market_type,
          self._account_request.exchange,
          self._symbol,
          ts)
    self._callbacks = []

  def _gen_common_symbol_stat_log(self) -> StrategyStatLog:
    assert self._account_request is not None
    assert self._symbol is not None
    assert self._accounting_currency is not None
    return StrategyStatLog(
        type=StrategyStatLog.SYMBOL,
        account_request=self._account_request,
        symbol=self._symbol,
        accounting_currency=self._accounting_currency)

  def _publish_strategy_stat_log(self, strat_stat_log: StrategyStatLog):
    assert strat_stat_log.strategy_stat.HasField('stat_ts')
    for callback in self._callbacks:
      callback(strat_stat_log.strategy_stat.stat_ts, strat_stat_log)

  def on_pnl_from_balance(self, accounting_currency: str, ts: int, value: float):
    strat_stat_log = StrategyStatLog(
        type=StrategyStatLog.STRATEGY,
        accounting_currency=accounting_currency,
        strategy_stat=StrategyStat(
            stat_ts=ts, pnl_mark_from_balance=value,))
    self._publish_strategy_stat_log(strat_stat_log)

  def on_pnl_from_fill(self, ts: int, value: float):
    strat_stat_log = self._gen_common_symbol_stat_log()
    strat_stat = StrategyStat(stat_ts=ts, pnl_mark_from_fill=value,)
    strat_stat_log.strategy_stat.MergeFrom(strat_stat)
    self._publish_strategy_stat_log(strat_stat_log)

  def on_position(self, ts: int, value: float):
    strat_stat_log = self._gen_common_symbol_stat_log()
    strat_stat = StrategyStat(stat_ts=ts, position=value,)
    strat_stat_log.strategy_stat.MergeFrom(strat_stat)
    self._publish_strategy_stat_log(strat_stat_log)

  def on_position_from_fill(self, ts: int, value: float):
    strat_stat_log = self._gen_common_symbol_stat_log()
    strat_stat = StrategyStat(stat_ts=ts, position_from_fill=value,)
    strat_stat_log.strategy_stat.MergeFrom(strat_stat)
    self._publish_strategy_stat_log(strat_stat_log)

  def register_strat_stat_observer(
      self, callback: Callable[[int, StrategyStatLog], None]):
    self._callbacks.append(callback)


class StrategyStatLogDumper(object):
  def __init__(self, writer: ProtoLogProvider):
    value_type = DefaultDict[str, List[StrategyStatLog]]
    self._strat_stats: value_type = defaultdict(list)
    self._symbol_stats: value_type = defaultdict(list)
    self._writer = writer
    self._pnl_stat_processor = PnlStatProcessor()

  @staticmethod
  def _validate_strategy_stat_log(log: StrategyStatLog):
    assert log.HasField('accounting_currency')
    assert log.strategy_stat.HasField('stat_ts')
    if log.type == StrategyStatLog.SYMBOL:
      assert log.account_request.HasField('market_type')
      assert log.account_request.HasField('exchange')
      assert log.account_request.HasField('owner')
      assert log.HasField('symbol')

  def _clear(self):
    self._strat_stats.clear()
    self._symbol_stats.clear()

  def _update_stats(self, log: StrategyStat, strat_stats: List[StrategyStat]):
    if len(strat_stats) == 0:
      strat_stats.append(log)
    else:
      ts = log.strategy_stat.stat_ts
      last_ts = strat_stats[-1].strategy_stat.stat_ts
      if ts > last_ts:
        strat_stats.append(log)
      elif ts == last_ts:
        strat_stats[-1].MergeFrom(log)
      else:
        raise AssertionError('%s vs %s' % (last_ts, ts))

  def on_strategy_stat_log(self, ts: int, log: StrategyStatLog):
    self._validate_strategy_stat_log(log)
    logs = self._pnl_stat_processor.process_strategy_stat_log(log)
    for each_log in logs:
      if each_log.type == StrategyStatLog.STRATEGY:
        strat_stats = self._strat_stats[each_log.accounting_currency]
        self._update_stats(each_log, strat_stats)
      if each_log.type == StrategyStatLog.SYMBOL:
        key = '%s.%s.%s.%s' % (
            each_log.account_request.market_type,
            each_log.account_request.exchange,
            each_log.account_request.owner,
            each_log.symbol)
        strat_stats = self._symbol_stats[key]
        self._update_stats(each_log, strat_stats)

  def write(self):
    logs = []
    for value in self._strat_stats.values():
      logs += value
    for value in self._symbol_stats.values():
      logs += value
    logs = sorted(logs, key=lambda x: x.strategy_stat.stat_ts)
    for log in logs:
      self._writer.write(log, flush=True, timestamp=log.strategy_stat.stat_ts)
    self._clear()


class PnlStatProcessor(object):
  def __init__(self):
    # symbol level
    self._symbol_pnl_from_fill: Dict[str, float] = {}
    # strat level
    self._strat_pnl_from_fill: Dict[str, float] = {}
    self._otc_pnl: Dict[str, float] = {}
    self._trading_pnl: Dict[str, float] = {}

  def _process_strategy_stat(self, log: StrategyStatLog
  ) -> List[StrategyStatLog]:
    processed = []
    accounting_ccy = log.accounting_currency
    if log.strategy_stat.HasField('pnl_mark_from_balance'):
      self._trading_pnl[accounting_ccy] = log.strategy_stat.pnl_mark_from_balance
      if not log.strategy_stat.HasField('pnl_mark'):
        trading_pnl_mark = log.strategy_stat.pnl_mark_from_balance
        otc_pnl_mark = 0
        if accounting_ccy in self._otc_pnl:
          otc_pnl_mark = self._otc_pnl[accounting_ccy]
          log.strategy_stat.otc_pnl_mark =  otc_pnl_mark
        log.strategy_stat.trading_pnl_mark = trading_pnl_mark
        log.strategy_stat.pnl_mark = trading_pnl_mark + otc_pnl_mark
    processed.append(log)
    return processed

  def _process_symbol_stat(self, log: StrategyStatLog
  )  -> List[StrategyStatLog]:
    processed = [log]
    accounting_ccy = log.accounting_currency
    if log.strategy_stat.HasField('pnl_mark_from_fill'):
      symbol_key = '%s.%s.%s.%s' % (
          log.account_request.market_type,
          log.account_request.exchange,
          log.account_request.owner,
          log.symbol,
      )
      symbol_pnl_from_fill = log.strategy_stat.pnl_mark_from_fill
      pnl_diff = symbol_pnl_from_fill - self._symbol_pnl_from_fill.get(symbol_key, 0)
      prev_strat_pnl_from_fill = self._strat_pnl_from_fill.get(accounting_ccy, 0)
      self._symbol_pnl_from_fill[symbol_key] = symbol_pnl_from_fill
      self._strat_pnl_from_fill[accounting_ccy] = prev_strat_pnl_from_fill + pnl_diff
      if log.account_request.exchange == 'Otc':
        prev_otc_pnl = self._otc_pnl.get(accounting_ccy, 0)
        self._otc_pnl[accounting_ccy] = prev_otc_pnl + pnl_diff
      new_log = StrategyStatLog(
          type=StrategyStatLog.STRATEGY,
          accounting_currency=accounting_ccy,
          strategy_stat=StrategyStat(
              stat_ts=log.strategy_stat.stat_ts,
              pnl_mark_from_fill=self._strat_pnl_from_fill[accounting_ccy]))
      processed.append(new_log)
    return processed

  def process_strategy_stat_log(self, orig_log: StrategyStatLog
  ) -> List[StrategyStatLog]:
    log = StrategyStatLog()
    log.CopyFrom(orig_log)
    if log.type == StrategyStatLog.STRATEGY:
      return self._process_strategy_stat(log)
    elif log.type == StrategyStatLog.SYMBOL:
      return self._process_symbol_stat(log)
    else:
      raise ValueError('Unknown log type: %s' % log.type)


class PassivePnlCalculator(object):
  def __init__(self,
               account_request: AccountRequestProto,
               symbol: str,
               ts: int,
               start_ts: Optional[int]=None,
               save_history: bool=False):
    self._calculator = PnlFromFillCalculator(
        account_request, symbol, ts, start_ts, save_pnl=save_history)

  @property
  def passive_pnl(self) -> MultiWindowStats:
    return self._calculator.pnl_from_fill

  def update_by_executor_config(self, ts: int, executor_config: ExecutorType):
    self._calculator.update_by_executor_config(ts, executor_config)

  def update_by_ticker(self, ts: int, ticker: ProductTicker):
    self._calculator.update_by_ticker(ts, ticker)

  def update_by_net_position(self, ts: int, position: float):
    if self._calculator.prev_pos_from_fill is None:
      self._calculator.update_by_net_position(ts, position)


class PnlFromFillCalculator(object):
  def __init__(self,
               account_request: AccountRequestProto,
               symbol: str,
               ts: int,
               start_ts: Optional[int]=None,
               save_pnl: bool=False,
               save_pos: bool=False):
    market_type = account_request.market_type
    exchange = account_request.exchange
    callback_type = Callable[[int, PnlFromFillCalculator], None]
    self._current_ts = ts
    self._start_ts = _get_today_timestamp(ts) if start_ts is None else start_ts
    self._account_request = account_request
    self._market_type = market_type
    self._exchange = exchange
    self._symbol = symbol
    self._product = _gen_product(market_type, exchange, symbol, ts)
    self._accounting_currency = _get_accounting_currency(
        market_type, exchange, symbol, ts)

    self._is_inverse = is_inverse_product(self._product)
    maker_fee, taker_fee = get_fee_rate(
        market_type, exchange, self._product, account_request.owner)
    self._maker_fee_rate = maker_fee
    self._taker_fee_rate = taker_fee

    self._pos_match_threshold = 30 * 10**9
    self._reserve: Optional[float] = None
    self._reserve_changed = False
    self._prev_fill_elem: Optional[Tuple[int, float]] = None
    self._prev_fill_price_adj: Optional[float] = None
    self._prev_pos_elem: Optional[Tuple[int, float]] = None
    self._prev_pos_from_fill: Optional[float] = None
    self._prev_pos_match_ts: Optional[int] = None
    self._pnl_from_fill = _create_multi_window_stats(self._current_ts,
                                                     start_ts=self._start_ts,
                                                     update_sum=False,
                                                     store_data=save_pnl,
                                                     stats_mark=True,
                                                     stats_4hr=True)
    self._pos_from_fill = _create_multi_window_stats(self._current_ts,
                                                     start_ts=self._start_ts,
                                                     update_sum=False,
                                                     store_data=save_pos,
                                                     stats_mark=True)
    self._pos_from_fill_adj = _create_multi_window_stats(self._current_ts,
                                                         start_ts=self._start_ts,
                                                         update_sum=False,
                                                         store_data=save_pos,
                                                         stats_mark=True)
    self._gross_pnl = 0.
    self._gross_pnl_adj = 0.
    self._last_fee = 0.
    self._total_fee = 0.
    self._net_pnl = 0.
    self._net_pnl_adj = 0.

    self._strat_stat_log_publisher = StrategyStatLogPublisher(
        ts=ts, account_request=account_request, symbol=symbol)
    self._pnl_from_fill.register_stats_mark_observer(self._on_pnl_from_fill)
    self._pos_from_fill.register_stats_mark_observer(
        self._strat_stat_log_publisher.on_position_from_fill)

  @property
  def accounting_currency(self) -> str:
    return self._accounting_currency

  @property
  def net_pnl_adj_mark(self) -> float:
    return self._pnl_from_fill.stats_mark.diff

  @property
  def net_pnl_adj(self) -> float:
    return self._net_pnl_adj

  @property
  def last_fee(self) -> float:
    return self._last_fee

  @property
  def pnl_from_fill(self) -> MultiWindowStats:
    return self._pnl_from_fill

  @property
  def pos_from_fill(self) -> MultiWindowStats:
    return self._pos_from_fill

  @property
  def pos_from_fill_adj(self) -> MultiWindowStats:
    return self._pos_from_fill_adj

  # self._prev_pos_from_fill is different from self._pos_from_fill.stats_mark.last
  # as self._prev_pos_from_fill is None when reserve changed
  @property
  def prev_pos_from_fill(self) -> Optional[float]:
    return self._prev_pos_from_fill

  @property
  def last_pos_from_fill_adj(self) -> Optional[float]:
    return self._pos_from_fill_adj.stats_mark.last

  def _on_pnl_from_fill(self, ts: int, value: float):
    self._strat_stat_log_publisher.on_pnl_from_fill(ts, self.net_pnl_adj_mark)

  def _update_prev_pos_match_ts(self, ts: int):
    assert ts is not None
    if self._prev_pos_elem is not None and \
       self.last_pos_from_fill_adj is not None and \
       _isclose(self._prev_pos_elem[1], self.last_pos_from_fill_adj):
      self._prev_pos_match_ts = ts

  def _update_pos_from_fill_adj(self, ts: int):
    if self._prev_pos_match_ts is None or \
       ts - self._prev_pos_match_ts > self._pos_match_threshold:
      if self._prev_pos_elem is not None:
        self._pos_from_fill_adj.update(*self._prev_pos_elem)

  def register_strat_stat_observer(
      self, cbk: Callable[[int, StrategyStatLog], None]):
    self._strat_stat_log_publisher.register_strat_stat_observer(cbk)

  def pos_from_fill_update_triggered(self, ts):
    return ts >= self._start_ts \
           and self._prev_pos_from_fill is None \
           and self._prev_pos_elem is not None

  def update_by_executor_config(self, ts: int, executor_config: ExecutorType):
    assert isinstance(executor_config, (PassiveExecutorConfig, AggressiveExecutorConfig))
    product = _gen_product(self._market_type, self._exchange, executor_config.symbol, ts)
    assert product.symbol == self._symbol, (product.symbol, self._symbol)
    if self._market_type == 'Spot':
      if executor_config.HasField('reserve'):
        reserve = executor_config.reserve
      else:
        reserve = 0.5 * (executor_config.min_pos + executor_config.max_pos)
    else:
      reserve = 0.
    if self._reserve is None:
      self._reserve = reserve
    else:
      if not _isclose(reserve, self._reserve):
        self._reserve_changed = True
        self._reserve = reserve
        self._prev_pos_from_fill = None

  def update_by_net_position(self, ts: int, position: float):
    self._update_prev_pos_match_ts(ts)
    if self.pos_from_fill_update_triggered(ts):
      if self._prev_pos_elem[0] < self._start_ts:
        pos_from_fill_elem = (self._start_ts, self._prev_pos_elem[1])
      else:
        pos_from_fill_elem = (ts, position)
      self._prev_pos_from_fill = pos_from_fill_elem[1]
      self._pos_from_fill.update(*pos_from_fill_elem)
      self._pos_from_fill_adj.update(*pos_from_fill_elem)
    self._reserve_changed = False
    self._prev_pos_elem = (ts, position)
    self._update_pos_from_fill_adj(ts)

  def update_by_order_event(self, ts: int, oe: OrderEvent):
    self._update_prev_pos_match_ts(ts)
    self._update_pos_from_fill_adj(ts)
    if oe.type == OrderEvent.ORDER_FILLED:
      self._update_by_fill(ts, oe)

  def update_by_ticker(self, ts: int, ticker: ProductTicker):
    assert isinstance(ticker, ProductTicker)
    assert ticker.symbol == self._symbol, (ticker.symbol, self._symbol)
    assert ticker.HasField('last')
    if self._prev_pos_elem is not None and self._prev_pos_elem[1] == 0:
      return

    # update on date change or every 10 mins
    if self._prev_fill_elem is None or \
       to_datetime(ts).date() != to_datetime(self._prev_fill_elem[0]).date() or \
       ts - self._prev_fill_elem[0] >= 10 * 60 * 10**9:
      oe = OrderEvent(type=OrderEvent.ORDER_FILLED,
                      event_time=ts,
                      symbol=ticker.symbol,
                      fill_price=ticker.last,
                      fill_qty=0)
      self.update_by_order_event(ts, oe)

  def _update_by_fill(self, ts: int, oe: OrderEvent):
    if _is_expired(self._product, ts):
      return
    if self.pos_from_fill_update_triggered(ts):
      if not self._reserve_changed:
        self._prev_pos_from_fill = self._prev_pos_elem[1]
        self._pos_from_fill.update(*self._prev_pos_elem)
        self._pos_from_fill_adj.update(*self._prev_pos_elem)

    assert oe.type == OrderEvent.ORDER_FILLED
    assert oe.HasField('event_time')
    assert oe.HasField('symbol')
    assert oe.HasField('fill_price')
    assert oe.HasField('fill_qty')
    if oe.fill_qty != 0:
      assert oe.order_side != order_enums.UNKNOWN_ORDER_SIDE

    sign = get_order_sign(oe.order_side)
    fill_price = oe.fill_price
    fill_qty = oe.fill_qty

    fill_price_adj = 1 / fill_price if self._is_inverse else fill_price
    prev_price_adj = self._prev_fill_price_adj
    multiplier = get_product_multiplier(self._market_type,
                                        self._exchange,
                                        self._symbol,
                                        current_datetime=datetime.datetime.utcfromtimestamp(
                                            ts / 10**9))
    inverse = -1 if self._is_inverse else 1

    fill_pq = fill_price_adj * fill_qty * multiplier
    is_maker = _is_maker(oe)
    fee_rate = self._maker_fee_rate if is_maker else self._taker_fee_rate
    fee = fee_rate * fill_pq
    if self._prev_pos_from_fill is not None:
      if prev_price_adj is not None:
        pnl = (fill_price_adj - prev_price_adj) * self._prev_pos_from_fill \
               * multiplier * inverse
        pnl_adj = (fill_price_adj - prev_price_adj) * self.last_pos_from_fill_adj \
                  * multiplier * inverse
        self._gross_pnl += pnl
        self._gross_pnl_adj += pnl_adj
      self._prev_pos_from_fill += fill_qty * sign
      self._pos_from_fill.update(ts, self._prev_pos_from_fill)
      self._pos_from_fill_adj.update(ts, self.last_pos_from_fill_adj + fill_qty * sign)

    self._last_fee = fee
    self._total_fee += fee
    self._net_pnl = self._gross_pnl - self._total_fee
    self._net_pnl_adj = self._gross_pnl_adj - self._total_fee
    self._prev_fill_elem = (ts, fill_price)
    self._prev_fill_price_adj = fill_price_adj
    self._pnl_from_fill.update(ts, elem=self._net_pnl_adj)


class SharpeRatioCalculator(object):
  def __init__(self, current_ts, start_ts, interval):
    self._sum = 0.
    self._square_sum = 0.
    self._count = 0
    self._current_ts = current_ts
    self._start_ts = start_ts
    self._interval = interval

  def update(self, ts, value):
    if ts <= self._current_ts or ts <= self._start_ts:
      return
    self._current_ts = ts
    self._count += 1
    self._square_sum += value ** 2
    self._sum += value

  @property
  def interval(self):
    return self._interval

  @property
  def sharpe(self):
    if self._count == 0:
      return None
    mean = self._sum / self._count
    variance = self._square_sum / self._count - mean ** 2
    if variance <= 0.:
      return None
    annualized_coef = math.sqrt(3600 * 24 * 365 / (self._interval / 10**9))
    return mean / math.sqrt(variance) * annualized_coef


class PositionStatsCalculator(object):
  def __init__(self,
               account_request: AccountRequestProto,
               symbol: str,
               current_ts: int,
               start_ts: int,
               save_history: bool=False):
    self._account_request = account_request
    self._market_type = account_request.market_type
    self._exchange = account_request.exchange
    self._symbol = symbol
    self._product = _gen_product(self._market_type, self._exchange, symbol, current_ts)
    self._current_ts = current_ts
    self._start_ts = start_ts
    # if position is different from zero from start_ts
    self._has_nonzero_position: bool = False
    self._single_side_pos_range: Optional[float] = None
    self._last_pos: Optional[ProductPosition] = None
    self._last_pos_upd_ts: Optional[int] = None
    elem_type = Optional[Tuple[int, float]]
    self._last_pos_elem: elem_type = None
    self._min_pos_elem: elem_type = None
    self._max_pos_elem: elem_type = None
    self._pos = _create_multi_window_stats(self._current_ts,
                                           start_ts=self._start_ts,
                                           update_sum=True,
                                           store_data=save_history,
                                           stats_mark=True,
                                           stats_4hr=True)
    self._pos_bias = _create_multi_window_stats(self._current_ts,
                                                start_ts=self._start_ts,
                                                update_sum=True,
                                                stats_mark=True)
    self._tw_pos = _create_multi_window_stats(self._current_ts,
                                              start_ts=self._start_ts,
                                              update_sum=True,
                                              stats_mark=True,
                                              stats_4hr=True)
    self._tw_abs_pos = _create_multi_window_stats(self._current_ts,
                                                  start_ts=self._start_ts,
                                                  update_sum=True,
                                                  stats_mark=True)
    self._tw_pos_bias = _create_multi_window_stats(self._current_ts,
                                                   start_ts=self._start_ts,
                                                   update_sum=True,
                                                   stats_mark=True)
    self._strat_stat_log_publisher = StrategyStatLogPublisher(
        ts=current_ts, account_request=self._account_request, symbol=self._symbol)
    self._pos.register_stats_mark_observer(
        self._strat_stat_log_publisher.on_position)

  @property
  def has_nonzero_position(self) -> bool:
    return self._has_nonzero_position

  @property
  def avg_position_mark(self) -> Optional[float]:
    result = self._calculate_twa_stats(self._tw_pos.stats_mark.sum,
                                       self._pos.stats_mark.first_elem,
                                       self._pos.stats_mark.last_elem)
    return result

  @property
  def avg_position_4hr(self) -> Optional[float]:
    result = self._calculate_twa_stats(self._tw_pos.stats_4hr.sum,
                                       self._pos.stats_4hr.first_elem,
                                       self._pos.stats_4hr.last_elem)
    return result

  @property
  def avg_position_bias_mark(self) -> Optional[float]:
    result = self._calculate_twa_stats(self._tw_pos_bias.stats_mark.sum,
                                       self._pos_bias.stats_mark.first_elem,
                                       self._pos_bias.stats_mark.last_elem)
    return result

  @property
  def net_position(self) -> MultiWindowStats:
    return self._pos

  @property
  def long_position(self) -> Optional[float]:
    long_position = None
    if self._last_pos is not None:
      long_position = self._last_pos.long_position
    return long_position

  @property
  def short_position(self) -> Optional[float]:
    short_position = None
    if self._last_pos is not None:
      short_position = self._last_pos.short_position
    return short_position

  @property
  def avg_abs_position_mark(self) -> Optional[float]:
    first_elem = self._pos.stats_mark.first_elem
    if first_elem is not None:
      first_ts, first_pos = self._pos.stats_mark.first_elem
      first_elem = (first_ts, abs(first_pos))
    last_elem = self._pos.stats_mark.last_elem
    if last_elem is not None:
      last_ts, last_pos = self._pos.stats_mark.last_elem
      last_elem = (last_ts, abs(last_pos))
    result = self._calculate_twa_stats(self._tw_abs_pos.stats_mark.sum,
                                       first_elem,
                                       last_elem)
    return result

  @property
  def total_abs_position(self) -> Optional[float]:
    total_abs_position = None
    if self._last_pos is not None:
      total_abs_position = abs(self._last_pos.long_position) + \
                           abs(self._last_pos.short_position)
    return total_abs_position

  def _calculate_twa_stats(self,
                           total: float,
                           first_elem: Optional[Tuple[int, float]],
                           last_elem: Optional[Tuple[int, float]]
  ) -> Optional[float]:
    result = None
    if first_elem is not None and last_elem is not None:
      last_ts, last_pos = last_elem
      first_ts, _ = first_elem
      assert self._current_ts >= last_ts >= first_ts, (
          self._current_ts, last_ts, first_ts)
      tw_pos = (self._current_ts - last_ts) * last_pos
      total += tw_pos
      if self._current_ts > first_ts:
        total_time = self._current_ts - first_ts
        result = total / total_time
    return result

  def register_strat_stat_observer(
      self, cbk: Callable[[int, StrategyStatLog], None]):
    self._strat_stat_log_publisher.register_strat_stat_observer(cbk)

  def update_timestamp(self, ts: int, start_ts: int):
    if ts > self._current_ts:
      self._current_ts = ts
    self._start_ts = start_ts
    for stats in (self._pos, self._pos_bias, self._tw_pos, self._tw_pos_bias):
      stats.update(ts, elem=None, start_ts=start_ts)

  def _update_min_max_position(self, ts: int, position: float):
    if ts >= self._start_ts:
      if self._min_pos_elem is None or position < self._min_pos_elem[1]:
        self._min_pos_elem = (ts, position)
      if self._max_pos_elem is None or position > self._max_pos_elem[1]:
        self._max_pos_elem = (ts, position)

  def update_by_position(self, ts: int, position: ProductPosition):
    if ts >= self._start_ts and position.net_position != 0:
      self._has_nonzero_position = True
    assert isinstance(position, ProductPosition)
    upd_pos = False
    if self._last_pos is None:
      upd_pos = True
    else:
      if not _isclose(self._last_pos.net_position, position.net_position, 10**-5):
        upd_pos = True
      elif ts - self._last_pos_upd_ts >= 10 * 60 * 10**9:
        upd_pos = True
    if upd_pos:
      self._update_by_net_position(ts, position.net_position)
      self._last_pos_upd_ts = ts
    self._last_pos = position

  def _update_by_net_position(self, ts: int, position: float):
    self._update_min_max_position(ts, position)
    self._pos.update(ts, elem=position)
    if self._last_pos_elem is not None:
      last_ts, last_pos = self._last_pos_elem
      last_abs_pos = abs(last_pos)
      tw_pos = (ts - last_ts) * last_pos
      tw_abs_pos = (ts - last_ts) * last_abs_pos
      self._tw_pos.update(last_ts, elem=tw_pos)
      self._tw_abs_pos.update(last_ts, elem=tw_abs_pos)
      if self._single_side_pos_range is not None and \
         self._single_side_pos_range != 0:
        pos_bias = position / self._single_side_pos_range
        tw_pos_bias = tw_pos / self._single_side_pos_range
        self._pos_bias.update(ts, elem=pos_bias)
        self._tw_pos_bias.update(last_ts, elem=tw_pos_bias)
    self._last_pos_elem = (ts, position)

  def update_by_executor_config(self, ts: int, executor_config: ExecutorType):
    assert isinstance(executor_config, (PassiveExecutorConfig, AggressiveExecutorConfig))
    product = _gen_product(self._market_type, self._exchange, executor_config.symbol, ts)
    assert product.symbol == self._symbol, (product.symbol, self._symbol)
    pos_range = executor_config.max_pos - executor_config.min_pos
    self._single_side_pos_range = pos_range / 2


class HourlyOrderPlotInfoContainer(object):
  def __init__(self):
    self._pnl = []
    self._position = []
    self._buy_order_price = []
    self._sell_order_price = []
    self._mid_price = []

  def get_order_plot_info(self):
    return {
        'pnl': self._pnl,
        'position': self._position,
        'buy_order_price': self._buy_order_price,
        'sell_order_price': self._sell_order_price,
        'mid_price': self._mid_price
    }

  def update_by_best_book_entry(self, ts, best_book_entry):
    assert best_book_entry.HasField('fetched_ts')
    if best_book_entry.bid0.HasField('price_float') and \
       best_book_entry.ask0.HasField('price_float'):
      mid_price = 0.5 * (best_book_entry.bid0.price_float + best_book_entry.ask0.price_float)
      self._mid_price.append((best_book_entry.fetched_ts, mid_price))

  def update_by_order_event(self, ts, order_event):
    if order_event.type == OrderEvent.ORDER_SUBMITTED:
      order_sign = get_order_sign(order_event.order_side)
      if order_sign == 1 and order_event.HasField('order_price'):
        self._buy_order_price.append((ts, order_event.order_price))
      if order_sign == -1 and order_event.HasField('order_price'):
        self._sell_order_price.append((ts, order_event.order_price))

  def update_by_pnl(self, ts, pnl):
    self._pnl.append((ts, pnl))

  def update_by_net_position(self, ts, position):
    self._position.append((ts, position))


class TradingStatsCalculator(object):
  def __init__(self,
               strategy_request: StrategyRequestProto,
               account_request: AccountRequestProto,
               symbol: str,
               ts: int,
               *,
               start_ts: Optional[int]=None,
               save_order_event: bool=False,
               save_history: bool=False,
               options_pricing_params: Optional[PricingParamsType]):
    market_type = account_request.market_type
    exchange = account_request.exchange
    self._current_ts = ts
    self._current_dt = datetime.datetime.utcfromtimestamp(ts / 10**9)
    self._start_ts = _get_today_timestamp(ts) if start_ts is None else start_ts
    self._last_order_ts = None
    self._last_fill_ts = None
    self._strategy_request = strategy_request
    self._account_request = account_request
    self._market_type = market_type
    self._exchange = exchange
    self._symbol = symbol
    self._product = _gen_product(market_type, exchange, symbol, ts)
    self._base = self._product.base.currency
    self._quote = self._product.quote.currency
    self._gen_product_dt = self._current_dt
    self._accounting_currency = _get_accounting_currency(
        market_type, exchange, symbol, ts)
    self._is_inverse = is_inverse_product(self._product)
    self._pnl_from_fill = PnlFromFillCalculator(account_request,
                                                symbol,
                                                ts,
                                                start_ts=self._start_ts,
                                                save_pnl=save_history,
                                                save_pos=save_history,)
    self._passive_pnl = PassivePnlCalculator(account_request,
                                             symbol,
                                             ts,
                                             start_ts=self._start_ts,
                                             save_history=save_history,)
    self._full_partial_fr = FullPartialFrCalculator(self._current_ts, start_ts=self._start_ts)
    self._last_ticker_ts = None
    self._ticker: Dict[str, float] = {}
    self._kline = {}
    self._options_pricing_params = \
        options_pricing_params if options_pricing_params is not None else {}
    self._hourly_plot_info_container = defaultdict(HourlyOrderPlotInfoContainer)

    self._net_pnl = self._pnl_from_fill.pnl_from_fill
    self._fee = _create_multi_window_stats(self._current_ts,
                                           start_ts=self._start_ts,
                                           update_sum=True,
                                           stats_mark=True)
    self._order_qty = _create_multi_window_stats(self._current_ts,
                                                 start_ts=self._start_ts,
                                                 update_sum=True,
                                                 stats_mark=True,
                                                 stats_4hr=True,
                                                 stats_8hr=True)
    self._fill_qty = _create_multi_window_stats(self._current_ts,
                                                start_ts=self._start_ts,
                                                update_sum=True,
                                                store_data=save_history,
                                                stats_mark=True,
                                                stats_4hr=True,
                                                stats_8hr=True)
    self._fill_pq = _create_multi_window_stats(self._current_ts,
                                               start_ts=self._start_ts,
                                               update_sum=True,
                                               stats_mark=True,
                                               stats_4hr=True,)

    self._position_stats = PositionStatsCalculator(account_request,
                                                   symbol,
                                                   self._current_ts,
                                                   self._start_ts,
                                                   save_history=save_history)
    self._last_pnl_from_fill = None
    self._sharpe_ratio = SharpeRatioCalculator(self._current_ts, self._start_ts, interval=60*10**9)

    self._oe_container = None
    if save_order_event:
      self._oe_container = SymbolOrderEventContainer(
          self._account_request, symbol, self._current_ts, self._start_ts)
    self._gen_fill_type_stats()

    self._ticker_log: Dict[Tuple[str, str], ExchangeTicker] = {}

  @property
  def accounting_currency(self) -> str:
    return self._accounting_currency

  @property
  def base(self) -> str:
    return self._base

  @property
  def quote(self) -> str:
    return self._quote

  @property
  def product(self) -> PyProduct:
    return self._product

  @property
  def market_type(self) -> str:
    return self._market_type

  @property
  def exchange(self) -> str:
    return self._exchange

  @property
  def last_net_position(self) -> Optional[float]:
    return self._position_stats.net_position.stats_mark.last

  @property
  def pnl_mark_from_fill(self) -> float:
    return self._net_pnl.stats_mark.diff

  @property
  def passive_trading_pnl_mark(self) -> float:
    return self._passive_pnl.passive_pnl.stats_mark.diff \
        if self._exchange != 'Otc' else 0

  @property
  def otc_pnl_mark(self) -> float:
    return self.pnl_mark_from_fill if self._exchange == 'Otc' else 0

  @property
  def pnl_4hr_from_fill(self) -> float:
    return self._net_pnl.stats_4hr.diff

  @property
  def trading_stats(self) -> StrategyStat:
    fill_ratio_mark, fill_ratio_4hr, fill_ratio_8hr = self._get_fill_ratio()
    full_fill_ratio_mark, partial_fill_ratio_mark, \
    full_fill_ratio_8hr, partial_fill_ratio_8hr = self._full_partial_fr.get_stats()

    fill_qty_mark = self._fill_qty.stats_mark.sum
    fill_qty_maker_mark = self._maker_qty.stats_mark.sum
    fill_qty_taker_mark = self._taker_qty.stats_mark.sum
    fill_qty_maker_inferred_mark = self._maker_qty_inferred.stats_mark.sum
    fill_qty_taker_inferred_mark = self._taker_qty_inferred.stats_mark.sum

    agg_fill_ratio = None
    agg_fill_ratio_inferred = None
    if not _isclose(fill_qty_mark, 0):
      agg_fill_ratio = fill_qty_taker_mark / fill_qty_mark
      agg_fill_ratio_inferred = fill_qty_taker_inferred_mark / fill_qty_mark

    symbol = '%s-USD' % self._accounting_currency
    ticker = self._get_ticker(symbol)
    passive_trading_pnl_mark = self.passive_trading_pnl_mark
    passive_trading_pnl_mark_in_usd = \
        ticker * passive_trading_pnl_mark if ticker is not None else None
    pnl_mark_from_fill = self.pnl_mark_from_fill
    pnl_mark_from_fill_in_usd = \
        ticker * pnl_mark_from_fill if ticker is not None else None
    otc_pnl_mark = self.otc_pnl_mark
    otc_pnl_mark_in_usd = ticker * otc_pnl_mark if ticker is not None else None
    pnl_4hr_from_fill = self.pnl_4hr_from_fill
    pnl_4hr_from_fill_in_usd = \
        ticker * pnl_4hr_from_fill if ticker is not None else None
    turnover_mark = self._fill_pq.stats_mark.sum
    turnover_maker_mark = self._maker_pq.stats_mark.sum
    turnover_taker_mark = self._taker_pq.stats_mark.sum
    turnover_maker_inferred_mark = self._maker_pq_inferred.stats_mark.sum
    turnover_taker_inferred_mark = self._taker_pq_inferred.stats_mark.sum
    turnover_mark_in_usd = self._get_turnover_mark_in_usd(
        fill_qty_mark, turnover_mark, ticker)
    turnover_maker_mark_in_usd = self._get_turnover_mark_in_usd(
        fill_qty_maker_mark, turnover_maker_mark, ticker)
    turnover_taker_mark_in_usd = self._get_turnover_mark_in_usd(
        fill_qty_taker_mark, turnover_taker_mark, ticker)
    turnover_maker_inferred_mark_in_usd = self._get_turnover_mark_in_usd(
        fill_qty_maker_inferred_mark, turnover_maker_inferred_mark, ticker)
    turnover_taker_inferred_mark_in_usd = self._get_turnover_mark_in_usd(
        fill_qty_taker_inferred_mark, turnover_taker_inferred_mark, ticker)

    market_share = None
    kline_key = (self._exchange, self._symbol, self._start_ts)
    kline = self._kline.get(kline_key, None)
    if kline is not None and kline.volume > 0:
      market_share = fill_qty_mark / kline.volume

    avg_position_mark = self._position_stats.avg_position_mark
    avg_position_mark_in_usd = self._get_position_in_usd(avg_position_mark)
    avg_abs_position_mark = self._position_stats.avg_abs_position_mark
    avg_abs_position_mark_in_usd = self._get_position_in_usd(avg_abs_position_mark)
    position = self.last_net_position
    total_abs_position = self._position_stats.total_abs_position
    position_in_usd = self._get_position_in_usd(position)
    total_abs_position_in_usd = self._get_position_in_usd(total_abs_position)
    if total_abs_position_in_usd is not None:
      # convert it to abs for products with negative delta like put options
      total_abs_position_in_usd = abs(total_abs_position_in_usd)
    greeks = OptionsGreeks()
    greeks_in_usd = OptionsGreeks()
    if self._market_type == 'Options':
      try:
        greeks = calculate_sabr_greeks(
            self._product,
            list(self._ticker_log.values()),
            [param.sabr for param in self._options_pricing_params.get(op_pb2.SABR, [])],
            to_datetime(self._current_ts))
      except:
        logging.error(traceback.format_exc())
      greeks_in_usd = self._get_greeks_in_usd(position, greeks)

    return StrategyStat(
        last_fill_ts=self._last_fill_ts,
        last_order_ts=self._last_order_ts,
        accounting_currency=self._accounting_currency,
        asset_name=self._product.base.currency,
        position=position,
        position_in_usd=position_in_usd,
        long_position=self._position_stats.long_position,
        short_position=self._position_stats.short_position,
        total_abs_position=total_abs_position,
        total_abs_position_in_usd=total_abs_position_in_usd,
        avg_position_mark=avg_position_mark,
        avg_position_4hr=self._position_stats.avg_position_4hr,
        avg_position_mark_in_usd=avg_position_mark_in_usd,
        avg_abs_position_mark=avg_abs_position_mark,
        avg_abs_position_mark_in_usd=avg_abs_position_mark_in_usd,
        avg_position_bias_mark=self._position_stats.avg_position_bias_mark,
        init_position_mark=self._position_stats.net_position.stats_mark.init,
        volume_mark=fill_qty_mark,
        volume_4hr=self._fill_qty.stats_4hr.sum,
        volume_maker_mark=fill_qty_maker_mark,
        volume_taker_mark=fill_qty_taker_mark,
        volume_maker_inferred_mark=fill_qty_maker_inferred_mark,
        volume_taker_inferred_mark=fill_qty_taker_inferred_mark,
        turnover_mark=turnover_mark,
        turnover_4hr=self._fill_pq.stats_4hr.sum,
        turnover_mark_in_usd=turnover_mark_in_usd,
        turnover_maker_mark=turnover_maker_mark,
        turnover_taker_mark=turnover_taker_mark,
        turnover_maker_inferred_mark=turnover_maker_inferred_mark,
        turnover_taker_inferred_mark=turnover_taker_inferred_mark,
        turnover_maker_mark_in_usd=turnover_maker_mark_in_usd,
        turnover_taker_mark_in_usd=turnover_taker_mark_in_usd,
        turnover_maker_inferred_mark_in_usd=turnover_maker_inferred_mark_in_usd,
        turnover_taker_inferred_mark_in_usd=turnover_taker_inferred_mark_in_usd,
        fill_ratio_mark=fill_ratio_mark,
        fill_ratio_8hr=fill_ratio_8hr,
        full_fill_ratio_mark=full_fill_ratio_mark,
        partial_fill_ratio_mark=partial_fill_ratio_mark,
        full_fill_ratio_8hr=full_fill_ratio_8hr,
        partial_fill_ratio_8hr=partial_fill_ratio_8hr,
        agg_ratio_qty_mark=agg_fill_ratio,
        agg_ratio_qty_inferred_mark=agg_fill_ratio_inferred,
        passive_trading_pnl_mark=passive_trading_pnl_mark,
        passive_trading_pnl_mark_in_usd=passive_trading_pnl_mark_in_usd,
        pnl_mark_from_fill=pnl_mark_from_fill,
        pnl_mark_from_fill_in_usd=pnl_mark_from_fill_in_usd,
        otc_pnl_mark=otc_pnl_mark,
        otc_pnl_mark_in_usd=otc_pnl_mark_in_usd,
        pnl_4hr_from_fill=pnl_4hr_from_fill,
        pnl_4hr_from_fill_in_usd=pnl_4hr_from_fill_in_usd,
        fee_mark=self._fee.stats_mark.sum,
        market_share=market_share,
        sharpe_ratio=self._sharpe_ratio.sharpe,
        options_model='SABR',
        implied_volatility=greeks.implied_volatility,
        delta=greeks.delta,
        gamma=greeks.gamma,
        vega=greeks.vega,
        theta=greeks.theta,
        delta_in_usd=greeks_in_usd.delta,
        gamma_in_usd=greeks_in_usd.gamma,
        vega_in_usd=greeks_in_usd.vega,
        theta_in_usd=greeks_in_usd.theta,)

  def get_stats_history(self):
    pnl_history = _multi_window_pnl_to_strategy_stat(self._net_pnl, 'from_fill')
    position_history = []
    position_from_fill_history = []
    for (ts, pos) in self._position_stats.net_position.stats_mark.history:
      each_position = [ProductPosition(symbol=self._symbol, net_position=pos)]
      account_position = AccountPosition(query_ts=ts, each_position=each_position)
      position_history.append(account_position)
    for (ts, pos) in self._pnl_from_fill.pos_from_fill.stats_mark.history:
      each_position = [ProductPosition(symbol=self._symbol, net_position=pos)]
      account_position = AccountPosition(query_ts=ts, each_position=each_position)
      position_from_fill_history.append(account_position)
    fill_history = [
        OrderEvent(event_time=ts, fill_qty=fill_qty, type=OrderEvent.ORDER_FILLED)
        for (ts, fill_qty) in self._fill_qty.stats_mark.history
    ]
    return {
        'pnl': pnl_history,
        'position': position_history,
        'position_from_fill': position_from_fill_history,
        'fill': fill_history
    }

  def get_order_plot_info(self, last=False):
    symbol_plot_msg_list = []
    if len(self._hourly_plot_info_container) > 0:
      last_hour = max(self._hourly_plot_info_container.keys())
      for hour, container in self._hourly_plot_info_container.items():
        order_plot_info = container.get_order_plot_info()
        if last and hour != last_hour:
          continue
        start_ts = self._start_ts + hour * 3600 * 10**9
        end_ts = start_ts + 3600 * 10**9
        symbol_plot_msg_list.append(
            SymbolOrderPlotMsg(account_request=self._account_request,
                               symbol=self._symbol,
                               pnl=order_plot_info['pnl'],
                               position=order_plot_info['position'],
                               buy_order_price=order_plot_info['buy_order_price'],
                               sell_order_price=order_plot_info['sell_order_price'],
                               mid_price=order_plot_info['mid_price'],
                               start_ts=start_ts,
                               end_ts=end_ts))
    return symbol_plot_msg_list

  def get_stat_log(self):
    stat_logs = []
    stat_log_common = StrategyStatLog(
        type=StrategyStatLog.SYMBOL,
        strategy=StrategyInfoProto(strategy_name=self._strategy_request.strategy_name),
        account_request=self._account_request,
        symbol=self._symbol,
        accounting_currency=self._accounting_currency)
    for ts, value in self._position_stats.net_position.stats_mark.history:
      stat_log = StrategyStatLog(strategy_stat=StrategyStat(stat_ts=ts, position=value))
      stat_log.MergeFrom(stat_log_common)
      stat_logs.append(stat_log)
    for ts, value in self._pnl_from_fill.pos_from_fill.stats_mark.history:
      stat_log = StrategyStatLog(
          strategy_stat=StrategyStat(stat_ts=ts, position_from_fill=value))
      stat_log.MergeFrom(stat_log_common)
      stat_logs.append(stat_log)
    for ts, value in self._net_pnl.stats_mark.history:
      stat_log = StrategyStatLog(
          strategy_stat=StrategyStat(stat_ts=ts, pnl_mark_from_fill=value))
      stat_log.MergeFrom(stat_log_common)
      stat_logs.append(stat_log)
    return stat_logs

  def get_log_history(self):
    order_event_history_df = pd.DataFrame(self._oe_container.get_history())
    if len(order_event_history_df) > 0:
      order_event_history_df.set_index('event_timestamp', inplace=True)
    position_df = pd.DataFrame(
        self._position_stats.net_position.stats_mark.history,
        columns=['event_timestamp', 'position'])
    position_df = position_df[
        position_df['position'].ne(position_df['position'].shift())].set_index(
            'event_timestamp')
    position_from_fill_df = pd.DataFrame(
        self._pnl_from_fill.pos_from_fill.stats_mark.history,
        columns=['event_timestamp', 'position_from_fill']).set_index('event_timestamp')
    pnl_from_fill_df = pd.DataFrame(
        self._net_pnl.stats_mark.history,
        columns=['event_timestamp', 'pnl_from_fill']).set_index('event_timestamp')
    output_df = pd.DataFrame()
    for df in (order_event_history_df, position_df, position_from_fill_df,
               pnl_from_fill_df):
      output_df = output_df.join(df, how='outer')
    return output_df.sort_index().reset_index().to_dict('records')

  def _get_pnl_from_fill_return(self, current_ts: int, pnl_from_fill: float
  ) -> Optional[float]:
    if self._last_pnl_from_fill is None:
      self._last_pnl_from_fill = (current_ts, pnl_from_fill,)
    elif current_ts >= self._last_pnl_from_fill[0] + self._sharpe_ratio.interval:
      pnl_from_fill_return = pnl_from_fill - self._last_pnl_from_fill[1]
      self._last_pnl_from_fill = (current_ts, pnl_from_fill,)
      return pnl_from_fill_return
    else:
      return None

  def _get_position_in_usd(self, position: Optional[float]
  ) -> Optional[float]:
    position_in_usd = None
    if position is None or _is_expired(self._product, self._current_ts):
      return position_in_usd
    try:
      account_request = AccountRequestProto(
        market_type=self._market_type, exchange=self._exchange)
      product_position = ProductPosition(symbol=self._symbol, net_position=position)
      currency, value = calculate_position_value(
          account_request,
          product_position,
          list(self._ticker_log.values()),
          options_pricing_params=self._options_pricing_params,
          current_datetime=datetime.datetime.utcfromtimestamp(self._current_ts / 10**9))
      ticker = self._get_ticker('%s-USD' % currency)
      position_in_usd = ticker * value if ticker is not None and value is not None \
                        else None
    except:
      if position != 0:
        logging.error(traceback.format_exc())
        logging.info(f'Fail to get position in USD: {self._symbol}, {position}')
    return position_in_usd

  def _get_greeks_in_usd(self,
                         position: Optional[float],
                         greeks: OptionsGreeks) -> OptionsGreeks:
    greeks_in_usd = OptionsGreeks()
    if position is None or _is_expired(self._product, self._current_ts):
      return greeks_in_usd
    try:
      account_request = AccountRequestProto(
          market_type=self._market_type, exchange=self._exchange)
      product_position = ProductPosition(symbol=self._symbol, net_position=1)
      currency, value = calculate_position_value(
          account_request,
          product_position,
          list(self._ticker_log.values()),
          options_pricing_params=self._options_pricing_params,
          current_datetime=datetime.datetime.utcfromtimestamp(self._current_ts / 10**9),
          disable_delta_adjust=True)
      qty_in_base = None
      if currency == self._base:
        qty_in_base = value
      else:
        price = self._get_ticker(f'{currency}-{self._base}')
        if price is not None:
          qty_in_base = value / price
      if self._quote == 'USD':
        quote_in_usd = 1
      else:
        quote_in_usd = self._get_ticker(f'{self._quote}-USD')
      gamma_in_usd = vega_in_usd = theta_in_usd = None
      if quote_in_usd is not None:
        if greeks.gamma_in_quote is not None and qty_in_base is not None:
          gamma_in_usd = greeks.gamma_in_quote * qty_in_base * position * quote_in_usd
        if greeks.theta_in_quote is not None:
          theta_in_usd = greeks.theta_in_quote * position * quote_in_usd
        if greeks.vega_in_quote is not None:
          vega_in_usd = greeks.vega_in_quote * position * quote_in_usd
        greeks_in_usd = OptionsGreeks(
            gamma=gamma_in_usd, theta=theta_in_usd, vega=vega_in_usd,)
    except:
      logging.error('Fail to get greeks in usd. %s' % self._product.symbol)
      logging.error(traceback.format_exc())
    return greeks_in_usd

  def _get_hourly_plot_info_container(self, ts: int) -> HourlyOrderPlotInfoContainer:
    hourly_plot_info = None
    if ts >= self._start_ts:
      hour_offset = int((ts - self._start_ts) / (3600 * 10**9))
      hourly_plot_info = self._hourly_plot_info_container[hour_offset]
    return hourly_plot_info

  def has_stats(self, ts: int) -> bool:
    has_stats = True
    if _isclose(self._order_qty.stats_8hr.sum, 0) and \
       _isclose(self._fill_qty.stats_8hr.sum, 0) and \
       _isclose(self._order_qty.stats_mark.sum, 0) and \
       _isclose(self._fill_qty.stats_mark.sum, 0) and \
       (self._market_type in ('Futures', 'Options') and
        not self._position_stats.has_nonzero_position):
      has_stats = False
    return has_stats

  def _get_ticker(self, symbol: str) -> Optional[float]:
    ticker = self._ticker.get(symbol, None)
    if ticker is not None and \
       self._current_ts - self._last_ticker_ts > TICKER_EXPIRE_TIME:
      ticker = None
    return ticker

  def _get_fill_ratio(self) -> Tuple[Optional[float],
                                            Optional[float],
                                            Optional[float]]:
    func = lambda fill, order: None if _isclose(order, 0) else fill / order
    fill_ratio_mark = func(self._fill_qty.stats_mark.sum, self._order_qty.stats_mark.sum)
    fill_ratio_4hr = func(self._fill_qty.stats_4hr.sum, self._order_qty.stats_4hr.sum)
    fill_ratio_8hr = func(self._fill_qty.stats_8hr.sum, self._order_qty.stats_8hr.sum)
    return fill_ratio_mark, fill_ratio_4hr, fill_ratio_8hr

  def _get_turnover_mark_in_usd(
      self, fill_qty: float, fill_pq: float, price_in_usd: Optional[float]
  ) -> Optional[float]:
    turnover_mark = calculate_turnover_in_usd(market_type=self._market_type,
                                              exchange=self._exchange,
                                              symbol=self._symbol,
                                              volume=fill_qty,
                                              turnover=fill_pq,
                                              turnover_price_in_usd=price_in_usd,
                                              current_datetime=self._gen_product_dt)
    return turnover_mark

  def _gen_fill_type_stats(self):
    self._maker_qty_inferred = _create_multi_window_stats(self._current_ts,
                                                          start_ts=self._start_ts,
                                                          update_sum=True,
                                                          stats_mark=True)
    self._taker_qty_inferred = _create_multi_window_stats(self._current_ts,
                                                          start_ts=self._start_ts,
                                                          update_sum=True,
                                                          stats_mark=True)
    self._maker_pq_inferred = _create_multi_window_stats(self._current_ts,
                                                         start_ts=self._start_ts,
                                                         update_sum=True,
                                                         stats_mark=True)
    self._taker_pq_inferred = _create_multi_window_stats(self._current_ts,
                                                         start_ts=self._start_ts,
                                                         update_sum=True,
                                                         stats_mark=True)
    self._maker_qty = _create_multi_window_stats(self._current_ts,
                                                 start_ts=self._start_ts,
                                                 update_sum=True,
                                                 stats_mark=True)
    self._taker_qty = _create_multi_window_stats(self._current_ts,
                                                 start_ts=self._start_ts,
                                                 update_sum=True,
                                                 stats_mark=True)
    self._maker_pq = _create_multi_window_stats(self._current_ts,
                                                start_ts=self._start_ts,
                                                update_sum=True,
                                                stats_mark=True)
    self._taker_pq = _create_multi_window_stats(self._current_ts,
                                                start_ts=self._start_ts,
                                                update_sum=True,
                                                stats_mark=True)

  def _update_fill_type_stats(self, ts: int, order_event: OrderEvent):
    fill_type = get_fill_type(order_event)
    fill_qty = order_event.fill_qty
    fill_pq = self._get_fill_pq(order_event)
    fill_qty_stat = None

    fill_pq_stat = None
    if fill_type == order_enums.UNKNOWN_FILL_TYPE:
      fill_type = get_fill_type_inferred(order_event)
      if fill_type == order_enums.MAKER_FILL_TYPE:
        fill_qty_stat = self._maker_qty_inferred
        fill_pq_stat = self._maker_pq_inferred
      elif fill_type == order_enums.TAKER_FILL_TYPE:
        fill_qty_stat = self._taker_qty_inferred
        fill_pq_stat = self._taker_pq_inferred
      else:
        pass
    else:
      if fill_type == order_enums.MAKER_FILL_TYPE:
        fill_qty_stat = self._maker_qty
        fill_pq_stat = self._maker_pq
      elif fill_type == order_enums.TAKER_FILL_TYPE:
        fill_qty_stat = self._taker_qty
        fill_pq_stat = self._taker_pq
      else:
        pass
    if fill_qty_stat is not None:
      fill_qty_stat.update(ts, elem=fill_qty)
    if fill_pq_stat is not None:
      fill_pq_stat.update(ts, elem=fill_pq)

  def _get_fill_pq(self, order_event: OrderEvent):
    fill_pq = calculate_turnover(market_type=self._market_type,
                                 exchange=self._exchange,
                                 symbol=self._symbol,
                                 volume=order_event.fill_qty,
                                 price=order_event.fill_price,
                                 current_datetime=self._current_dt)
    return fill_pq

  def update_position(self, ts: int, position: ProductPosition):
    assert isinstance(position, ProductPosition)
    # only update by different position to reduce latency and memory
    net_position = position.net_position
    if self.last_net_position is None or \
       not _isclose(net_position, self.last_net_position, 10**-5):
      self._pnl_from_fill.update_by_net_position(ts, net_position)
      self._passive_pnl.update_by_net_position(ts, net_position)
      container = self._get_hourly_plot_info_container(ts)
      if container is not None:
        container.update_by_net_position(ts, net_position)
    # self.update_by_positioins() changes self.last_net_position,
    # so it must be executed after isclose if statement
    self._position_stats.update_by_position(ts, position)

  def update_ticker(self, ts: int, ticker: Dict[str, float]):
    if self._last_ticker_ts is None or ts > self._last_ticker_ts:
      self._last_ticker_ts = ts
      self._ticker = ticker

  def update_kline(self, kline: List[DBOutputKline]):
    kline_info = {}
    for entry in kline:
      if entry is None:
        continue
      product_kline = entry.klines
      assert isinstance(product_kline, ProductKline)
      for kline in product_kline.klines:
        key = (product_kline.exchange, product_kline.symbol, kline.kline_timestamp)
        kline_info[key] = kline
    self._kline = kline_info

  def update_options_pricing_params(self, options_pricing_param: PricingParamsType):
    self._options_pricing_params = options_pricing_param

  def update_by_ticker(self, ts: int, ticker: ExchangeTicker):
    self._ticker_log[(ticker.market_type, ticker.exchange)] = ticker

  def update_by_order_event(self, ts: int, order_event: OrderEvent):
    assert isinstance(order_event, OrderEvent)
    self._full_partial_fr.update_by_order_event(ts, order_event)
    if self._oe_container is not None:
      self._oe_container.update_by_order_event(ts, order_event)
    container = self._get_hourly_plot_info_container(ts)
    self._pnl_from_fill.update_by_order_event(ts, order_event)
    if container is not None:
      container.update_by_order_event(ts, order_event)
    if order_event.type == OrderEvent.ORDER_SUBMITTED:
      self._last_order_ts = ts
      self._order_qty.update(ts, elem=order_event.order_qty)
    elif order_event.type == OrderEvent.ORDER_FILLED:
      self._last_fill_ts = ts
      net_pnl_from_fill = self._pnl_from_fill.net_pnl_adj_mark

      pnl_from_fill_return = self._get_pnl_from_fill_return(ts, net_pnl_from_fill)
      if pnl_from_fill_return is not None:
        self._sharpe_ratio.update(ts, pnl_from_fill_return)
      if container is not None:
        container.update_by_pnl(ts, net_pnl_from_fill)
      self._fee.update(ts, elem=self._pnl_from_fill.last_fee)
      self._fill_qty.update(ts, elem=order_event.fill_qty)
      fill_pq = self._get_fill_pq(order_event)
      self._fill_pq.update(ts, elem=fill_pq)
      self._update_fill_type_stats(ts, order_event)
    else:
      pass

  def update_by_executor_config(self, ts: int, executor_config: ExecutorType):
    self._pnl_from_fill.update_by_executor_config(ts, executor_config)
    self._passive_pnl.update_by_executor_config(ts, executor_config)
    self._position_stats.update_by_executor_config(ts, executor_config)

  def update_by_best_book_entry(self, ts: int, best_book_entry: BookEntry):
    assert best_book_entry.symbol == self._symbol, (best_book_entry.symbol, self._symbol)
    container = self._get_hourly_plot_info_container(ts)
    if container is not None:
      container.update_by_best_book_entry(ts, best_book_entry)

  def update_timestamp(self, ts: int, start_ts: int):
    if ts > self._current_ts:
      self._current_ts = ts
      self._current_dt = datetime.datetime.utcfromtimestamp(ts / 10**9)
    self._start_ts = start_ts
    for stats in (self._order_qty, self._fill_qty, self._fill_pq):
      stats.update(ts, elem=None, start_ts=start_ts)
    self._position_stats.update_timestamp(ts, start_ts)


class SingleOrderContext(object):
  def __init__(self):
    self.pending_time = None
    self.accepted_time = None
    self.dead_time = None
    self.order_state = order_enums.CREATED_ORDER


class SymbolOrderEventContainer(object):
  def __init__(self, account_request, symbol, ts, start_ts):
    self._account_request = account_request
    self._symbol = symbol
    self._start_ts = start_ts
    self._working_orders = set()
    self._accm_num_error_orders = 0
    self._order_events = []
    self._order_contexts = {}

  def _gen_order_dict(self, ts, order_event):
    assert order_event.HasField('event_time')
    assert order_event.HasField('type')
    assert order_event.HasField('order_state')
    order_sign = None
    if order_event.HasField('order_side'):
      order_sign = 'BUY_ORDER' if get_order_sign(order_event.order_side) > 0 \
                   else 'SELL_ORDER'
    order_dict_orig = json_format.MessageToDict(
        order_event, preserving_proto_field_name=True)
    keys = [
        'type', 'order_price', 'order_qty', 'order_side', 'internal_order_id',
        'external_order_id', 'fill_price', 'fill_qty', 'order_state']
    order_dict = {key: order_dict_orig.get(key) for key in keys}
    # proto field with type of int64 is converted to string in MessageToDict()
    order_dict['event_timestamp'] = int(ts)
    internal_order_id = order_dict['internal_order_id']
    if order_event.HasField('proc_order_id'):
      assert internal_order_id is None
      internal_order_id = str(order_event.proc_order_id)
    assert internal_order_id is not None
    order_dict['internal_order_id'] = internal_order_id
    order_dict['order_sign'] = order_sign
    return order_dict

  def _get_single_order_context(self, internal_order_id):
    if internal_order_id not in self._order_contexts:
      self._order_contexts[internal_order_id] = SingleOrderContext()
    return self._order_contexts[internal_order_id]

  def update_by_order_event(self, ts, order_event):
    if ts < self._start_ts:
      return
    prev_num_working_orders = len(self._working_orders)
    order_dict = self._gen_order_dict(ts, order_event)
    internal_order_id = order_dict['internal_order_id']
    order_context = self._get_single_order_context(internal_order_id)
    if order_event.order_state == order_enums.PENDING_ORDER:
      assert order_context.pending_time is None
      assert order_context.accepted_time is None
      assert order_context.dead_time is None
      assert order_context.order_state == order_enums.CREATED_ORDER
      order_context.pending_time = ts
      self._working_orders.add(internal_order_id)
    elif order_event.order_state == order_enums.WORKING_ORDER:
      assert order_context.order_state in (
          order_enums.CREATED_ORDER,
          order_enums.PENDING_ORDER,
          order_enums.WORKING_ORDER), order_context.order_state
      if order_context.order_state == order_enums.CREATED_ORDER:
        logging.debug(
            'working order not found in history. %s %s' % (ts, order_event))
      if order_context.order_state == order_enums.PENDING_ORDER:
        assert order_context.pending_time is not None
      if order_context.accepted_time is None:
        order_context.accepted_time = ts
        if order_context.pending_time is not None:
          order_dict['order_pending_duration'] = \
              order_context.accepted_time - order_context.pending_time
      else:
        assert order_context.accepted_time <= ts
      self._working_orders.add(internal_order_id)
    elif order_event.order_state == order_enums.DEAD_ORDER:
      if order_context.order_state == order_enums.CREATED_ORDER:
        logging.debug(
            'dead order not found in history. %s %s' % (ts, order_event))
      if order_context.dead_time is None:
        order_context.dead_time = ts
      else:
        assert order_context.dead_time <= ts
      if order_context.pending_time is not None:
        order_dict['order_life_duration'] = \
            order_context.dead_time - order_context.pending_time
      self._working_orders.discard(internal_order_id)
    else:
      raise ValueError('Unknown order state. %s' % order_event.order_state)
    order_context.order_state = order_event.order_state
    curr_num_working_orders = len(self._working_orders)
    if curr_num_working_orders != prev_num_working_orders:
      order_dict['num_working_orders'] = curr_num_working_orders
      order_dict['working_orders'] = ','.join(sorted(list(self._working_orders)))
    if order_event.type in (OrderEvent.ORDER_ERROR, OrderEvent.ORDER_REJECTED):
      self._accm_num_error_orders += 1
      order_dict['accm_num_error_orders'] = self._accm_num_error_orders
    if order_context.pending_time is not None:
      order_dict['pending_timestamp'] = order_context.pending_time
    self._order_events.append(order_dict)

  def get_history(self):
    return self._order_events


class PnlFromBalanceCalculator(object):
  def __init__(self, *,
               strategy_request: StrategyRequestProto,
               currency: str,
               ts: int,
               start_ts: int,
               reserve: DefaultDict[AccountKeyType, Dict[str, float]],
               reserve_adjust_type,
               pnl_balance_details: List[AccountBalance]=None,
               ticker: Optional[Dict[str, float]]=None,
               save_history: bool=False):
    self._current_ts = ts
    self._strategy_request = strategy_request
    self._currency = currency
    self._reserve = copy.deepcopy(reserve)
    self._update_adj = False
    # there may be multiple reserve changes between pnl balance
    self._reserve_diff_details: \
        DefaultDict[AccountKeyType, List[Optional[float]]] = defaultdict(list)
    self._pnl_balance = _create_multi_window_stats(
        ts, start_ts=start_ts, update_sum=False, store_data=save_history,
        stats_mark=True, stats_4hr=True)
    self._pnl_balance_adj = _create_multi_window_stats(
        ts, start_ts=start_ts, update_sum=True, store_data=save_history,
        stats_mark=True, stats_4hr=True)
    self._pnl_adj_info = None
    self._pnl_adjusted_date = set()
    self._prev_pnl = None
    self._ticker = ticker if ticker is not None else {}
    self._last_ticker_ts = None
    self._reserve_adjust_type = reserve_adjust_type
    pnl_bal_details_type = Dict[AccountKeyType, float]
    self._prev_pnl_bal_details: Optional[pnl_bal_details_type] = None
    pnl_bal_dict = None
    if pnl_balance_details is not None:
       pnl_bal_dict = self._get_pnl_balance_dict(pnl_balance_details)
    self._curr_pnl_bal_details: Optional[pnl_bal_details_type] = pnl_bal_dict

    self._strat_stat_log_publisher = StrategyStatLogPublisher()
    self._pnl_balance.register_stats_mark_observer(self._on_pnl_balance)

  @property
  def pnl_mark(self) -> float:
    pnl = self._pnl_balance.stats_mark.diff
    return pnl

  @property
  def pnl_4hr(self) -> float:
    pnl = self._pnl_balance.stats_4hr.diff
    return pnl

  @property
  def pnl_balance(self) -> MultiWindowStats:
    return self._pnl_balance

  def _clear_reserve_diff_details(self):
    self._reserve_diff_details.clear()

  def _save_prev_pnl_balance(self):
    self._prev_pnl = [self._pnl_balance.stats_mark.diff,
                      self._pnl_balance.stats_4hr.diff]

  def _get_pnl_balance_dict(self, pnl_balance_details: List[AccountBalance]):
    pnl_bal_dict = {}
    for acct_bal in pnl_balance_details:
      acct_key = (acct_bal.market_type, acct_bal.exchange, acct_bal.owner)
      for ccy_bal in acct_bal.each_balance:
        if ccy_bal.currency == self._currency:
          pnl_bal_dict[acct_key] = ccy_bal.total
    return pnl_bal_dict

  def _get_ticker(self, symbol: str) -> Optional[float]:
    ticker = self._ticker.get(symbol, None)
    if ticker is None:
      try:
        base, quote = symbol.split('-')
        if base == quote:
          ticker = 1
        else:
          ticker = self._ticker.get(f'{base}-USD') / self._ticker.get(f'{quote}-USD')
      except Exception:
        pass
    # muted this logic as historical feed data is not available.
    """
    if ticker is not None and \
       self._current_ts - self._last_ticker_ts > TICKER_EXPIRE_TIME:
      ticker = None
    """
    return ticker

  def _get_reserve_adj_value(
      self, ts: int, pnl_balance_diff: float, reserve_diff: Optional[float]):
    adj_value = pnl_balance_diff
    strategy_name = self._strategy_request.strategy_name
    trading_date = datetime.datetime.utcfromtimestamp(ts / 10**9).date()
    if reserve_diff is not None:
      if self._reserve_adjust_type == strategy_reporter_proto.RESERVE_DIFF_VALUE:
        if not _adjust_by_pnl_balance_diff(strategy_name, trading_date):
          adj_value = reserve_diff
      else:
        if _adjust_by_reserve_diff(strategy_name, trading_date):
          adj_value = reserve_diff
    return adj_value

  def _get_pnl_adjust_value(self, *, ts: int, curr_pnl: float, prev_pnl: float):
    ret = None
    if self._pnl_adj_info is not None:
      td_str = datetime.datetime.utcfromtimestamp(ts / 10**9).date().strftime('%Y%m%d')
      if td_str not in self._pnl_adjusted_date:
        adj_entries = self._pnl_adj_info.get(td_str, [])
        adj_values = [
            entry['pnl_adj']
            for entry in adj_entries
            if self._strategy_request.strategy_name == entry['strategy_name']
            and self._currency == entry['accounting_currency']
        ]
        if len(adj_values) > 0:
          assert len(adj_values) == 1, len(adj_values)
          adj_value = adj_values[0]
          if 0.95 < (prev_pnl - curr_pnl) / adj_value < 1.05:
            ret = adj_value
            self._pnl_adjusted_date.add(td_str)
            logging.debug(
                'td: %s, strategy: %s, currency: %s, pnl_adj: %s, ts: %s, curr_pnl: %s, prev_pnl: %s'
                % (td_str,
                   self._strategy_request.strategy_name,
                   self._currency,
                   adj_value,
                   ts,
                   curr_pnl,
                   prev_pnl))
    return ret

  def _on_pnl_balance(self, ts: int, value: float):
    self._strat_stat_log_publisher.on_pnl_from_balance(self._currency, ts, self.pnl_mark)

  def _update_for_pnl_adjust_info(self):
    if self._prev_pnl is not None:
      pnl_stats = [self._pnl_balance.stats_mark,
                   self._pnl_balance.stats_4hr]
      pnl_adj_stats = [self._pnl_balance_adj.stats_mark,
                       self._pnl_balance_adj.stats_4hr]
      assert len(self._prev_pnl) == len(pnl_stats) == len(pnl_adj_stats)
      for prev_pnl, stat, stat_adj in zip(self._prev_pnl, pnl_stats, pnl_adj_stats):
        last_elem = stat.last_elem
        if last_elem is None:
          continue
        ts = last_elem[0]
        pnl_balance = last_elem[1]
        curr_pnl = stat.diff
        pnl_adj = self._get_pnl_adjust_value(ts=ts, curr_pnl=curr_pnl, prev_pnl=prev_pnl)
        if pnl_adj is not None:
          stat_adj.update(ts, elem=pnl_adj)
          pnl_balance += pnl_adj
          stat.pop()
          stat.update(ts, elem=pnl_balance)

  def register_strat_stat_observer(
      self, cbk: Callable[[int, StrategyStatLog], None]):
    self._strat_stat_log_publisher.register_strat_stat_observer(cbk)

  def update_by_strat_reporter_config(
      self, ts: int, reporter_config: strategy_reporter_proto.StrategyReporterConfig):
    self._reserve_adjust_type = reporter_config.reserve_adjust_type

  def _update_by_pnl_balance_wo_details(self, ts: int, pnl_balance: CurrencyBalance):
    assert isinstance(pnl_balance, CurrencyBalance)
    assert pnl_balance.currency == self._currency
    assert pnl_balance.HasField('total')
    pnl_bal_value = pnl_balance.total
    self._save_prev_pnl_balance()
    reserve_diff = None
    reserve_diff_values = []
    for acct_key, diff_values in self._reserve_diff_details.items():
      reserve_diff_values.extend(diff_values)
    if len(reserve_diff_values) > 0 and \
       all([i is not None for i in reserve_diff_values]):
      reserve_diff = sum(reserve_diff_values)
    self._clear_reserve_diff_details()
    pnl_stats = [self._pnl_balance.stats_mark, self._pnl_balance.stats_4hr]
    pnl_adj_stats = [self._pnl_balance_adj.stats_mark, self._pnl_balance_adj.stats_4hr]
    for stat, stat_adj in zip(pnl_stats, pnl_adj_stats):
      if self._update_adj:
        if stat.last is not None:
          pnl_balance_diff = stat.last - (pnl_bal_value + stat_adj.sum)
          adj = self._get_reserve_adj_value(ts, pnl_balance_diff, reserve_diff)
          stat_adj.update(ts, elem=adj)
          logging.info(
              f'strategy: {self._strategy_request.strategy_name}, ' + \
              f'ts: {ts}, ' + \
              f'time: {to_datetime(ts)}, ' + \
              f'reserve_diff: {reserve_diff}, ' + \
              f'pnl_bal_diff: {pnl_balance_diff}, ' + \
              f'revision_adj: {adj}.'
          )
      adj_pnl_bal_value = pnl_bal_value + stat_adj.sum
      """
      if stat.last is not None and stat._start_ts is not None:
        pnl_diff = adj_pnl_bal_value - stat.last
        logging.debug('pnl diff: %s, %s' % (
            datetime.datetime.utcfromtimestamp(ts / 10**9), adj_pnl_bal_value - stat.last))
      """
      stat.update(ts, elem=adj_pnl_bal_value)
    self._update_adj = False
    self._update_for_pnl_adjust_info()

  def _get_new_accts_diff(self) -> Dict[AccountKeyType, float]:
    new_accts_diff = {}
    if self._prev_pnl_bal_details is not None and \
       self._curr_pnl_bal_details is not None:
      for acct_key, pnl_bal_value in self._curr_pnl_bal_details.items():
        if acct_key not in self._prev_pnl_bal_details:
          new_accts_diff[acct_key] = pnl_bal_value
    return new_accts_diff

  def _get_removed_accts_diff(self) -> Dict[AccountKeyType, float]:
    removed_accts_diff = {}
    if self._prev_pnl_bal_details is not None and \
       self._curr_pnl_bal_details is not None:
      for acct_key, pnl_bal_value in self._prev_pnl_bal_details.items():
        if acct_key not in self._curr_pnl_bal_details:
          removed_accts_diff[acct_key] = pnl_bal_value
    return removed_accts_diff

  def _get_reserve_diff(self) -> Dict[AccountKeyType, Optional[float]]:
    reserve_diff = {}
    for acct_key, diff_values in self._reserve_diff_details.items():
      if len(diff_values) > 0:
        diff = None
        if all([elem is not None for elem in diff_values]):
          diff = sum(diff_values)
        reserve_diff[acct_key] = diff
    return reserve_diff

  def _get_pnl_balance_diff(self) -> Dict[AccountKeyType, float]:
    pnl_bal_diff = {}
    for acct_key in self._reserve_diff_details.keys():
      prev_value = curr_value = None
      if self._prev_pnl_bal_details is not None:
        prev_value = self._prev_pnl_bal_details.get(acct_key)
      if self._curr_pnl_bal_details is not None:
        curr_value = self._curr_pnl_bal_details.get(acct_key)
      if prev_value is not None and curr_value is not None:
        pnl_bal_diff[acct_key] = prev_value - curr_value
    return pnl_bal_diff

  def _update_by_pnl_balance_with_details(
      self, ts: int, pnl_balance: CurrencyBalance, is_fake: bool=False):
    assert isinstance(pnl_balance, CurrencyBalance)
    assert pnl_balance.currency == self._currency
    if is_fake:
      pnl_bal_value = 0
    else:
      assert pnl_balance.HasField('total')
      pnl_bal_value = pnl_balance.total
    self._save_prev_pnl_balance()
    new_accts_diff = self._get_new_accts_diff()
    removed_accts_diff = self._get_removed_accts_diff()
    reserve_diff = self._get_reserve_diff()
    pnl_balance_diff = self._get_pnl_balance_diff()
    revision_diff = {}
    for acct_key in reserve_diff.keys():
      if acct_key in pnl_balance_diff:
        adj = self._get_reserve_adj_value(
            ts, pnl_balance_diff[acct_key], reserve_diff[acct_key])
        revision_diff[acct_key] = adj
    new_accts_adj = -1 * sum([diff for _, diff in new_accts_diff.items()])
    removed_accts_adj = sum([diff for _, diff in removed_accts_diff.items()])
    reserve_adj = sum([diff for _, diff in revision_diff.items()])
    total_adj = new_accts_adj + removed_accts_adj + reserve_adj

    pnl_stats = [self._pnl_balance.stats_mark, self._pnl_balance.stats_4hr]
    pnl_adj_stats = [self._pnl_balance_adj.stats_mark, self._pnl_balance_adj.stats_4hr]
    for stat, stat_adj in zip(pnl_stats, pnl_adj_stats):
      if self._update_adj or new_accts_adj != 0 or removed_accts_adj != 0:
        if stat.last is not None:
          stat_adj.update(ts, elem=total_adj)
          logging.info(
              f"strategy: {self._strategy_request.strategy_name}, " + \
              f"ts: {ts}, " + \
              f"time: {to_datetime(ts)}, " + \
              f"new_accts_diff: {new_accts_diff}, " + \
              f"removed_accts_diff: {removed_accts_diff}, " + \
              f"reserve_diff: {reserve_diff}, " + \
              f"pnl_balance_diff: {pnl_balance_diff}, " + \
              f"revision_diff: {revision_diff}, " + \
              f"new_accts_adj: {new_accts_adj}, " + \
              f"removed_accts_adj: {removed_accts_adj}, " + \
              f"reserve_adj: {reserve_adj}, " + \
              f"total_adj: {total_adj}"
          )
      adj_pnl_bal_value = pnl_bal_value + stat_adj.sum
      stat.update(ts, elem=adj_pnl_bal_value)
    self._update_adj = False
    self._clear_reserve_diff_details()

  def update_by_pnl_balance_details(self,
                                    ts: int,
                                    pnl_bal_details: List[AccountBalance]):

    if len(pnl_bal_details) == 0:
      return
    pnl_bal_dict = self._get_pnl_balance_dict(pnl_bal_details)
    self._prev_pnl_bal_details = self._curr_pnl_bal_details
    self._curr_pnl_bal_details = pnl_bal_dict
    # when a acct is removed, it's possible that no following pnl balance is published,
    # so need to create a fake one to handle removed acct case.
    if len(pnl_bal_dict) == 0:
      self._update_by_pnl_balance_with_details(
          ts, CurrencyBalance(currency=self._currency), True)

  def update_by_pnl_balance(self, ts: int, pb: StrategyLog):
    assert pb.HasField('pnl_balance')
    pnl_balance = pb.pnl_balance
    if self._prev_pnl_bal_details is not None and \
       self._curr_pnl_bal_details is not None:
      self._update_by_pnl_balance_with_details(ts, pnl_balance)
    else:
      self._update_by_pnl_balance_wo_details(ts, pnl_balance)

  def update_by_reserve(self,
                        ts: int,
                        reserve: AccountBalance,
                        account_request: AccountRequestProto):
    self._current_ts = ts
    assert isinstance(reserve, AccountBalance)
    acct_key = _get_account_key(account_request)
    reserve = _get_reserve_dict(reserve)
    prev_reserve = self._reserve.get(acct_key)
    self._reserve[acct_key] = reserve
    if self._curr_pnl_bal_details is not None:
      if acct_key not in self._curr_pnl_bal_details:
        return
    if prev_reserve is not None and reserve != prev_reserve:
      self._update_adj = True
      price_not_found = False
      reserve_not_found = False
      all_coins = list(set(list(prev_reserve.keys()) + list(reserve.keys())))
      for coin in all_coins:
        if coin not in prev_reserve or coin not in reserve:
          reserve_not_found = True
          logging.debug(f'reserve not found: {coin}')
      reserve_df = pd.DataFrame([prev_reserve, reserve], index=['prev', 'curr'])
      reserve_df = reserve_df[sorted(reserve_df.columns)]
      reserve_df.loc['diff'] = reserve_df.loc['curr'].subtract(
          reserve_df.loc['prev'], fill_value=0)
      price_dict = {}
      for coin in reserve_df.columns:
        symbol = f'{coin}-{self._currency}'
        price = 1 if coin == self._currency else self._get_ticker(symbol)
        if price is None:
          price_not_found = True
          logging.info(f'price not found: {symbol}')
        price_dict[coin] = price
      price_df = pd.DataFrame([price_dict], index=['price'])
      reserve_df = reserve_df.append(price_df)
      reserve_df.loc['diff_value'] = reserve_df.loc['diff'].multiply(
          reserve_df.loc['price'])
      total_diff_value = reserve_df.loc['diff_value'].sum()
      if reserve_not_found or price_not_found:
        total_diff_value = None
      self._reserve_diff_details[acct_key].append(total_diff_value)
      reserve_changed_ccy = reserve_df.columns[
          reserve_df.loc['prev'] != reserve_df.loc['curr']]
      reserve_diff_df = reserve_df.loc[:, reserve_changed_ccy]
      logging.info(
          f'reserve changed. ' + \
          f'strategy: {self._strategy_request.strategy_name}, ' + \
          f'account: {account_request}, ' + \
          f'accounting_ccy: {self._currency}, ' + \
          f'ts: {ts}, ' + \
          f'time: {to_datetime(ts)}, ' + \
          f'diff_value: {total_diff_value}, ' + \
          f'reserve_diff: \n{reserve_diff_df.to_string()}.'
      )

  def update_by_ticker(self, ts: int, ticker: ExchangeTicker):
    if self._last_ticker_ts is None or ts > self._last_ticker_ts:
      self._last_ticker_ts = ts
      for elem in ticker.each_ticker:
        self._ticker[elem.symbol] = elem.last

  def update_pnl_adjust_info(self, pnl_adjust_info):
    self._pnl_adj_info = pnl_adjust_info

  def update_timestamp(self, ts: int, start_ts: int):
    for stats in (self._pnl_balance, self._pnl_balance_adj):
      stats.update(ts, elem=None, start_ts=start_ts)


class StrategySummaryCalculator(object):
  def __init__(self,
               *,
               strategy_name: str,
               machine: str,
               source_info: str,
               current_timestamp: Optional[int]=None,
               start_timestamp: Optional[int]=None,
               daily_reset_disabled: bool=False,
               save_history: bool=False,
               save_order_event: bool=False,
               strat_stat_log_dir: str=None):
    self._machine = machine
    self._strategy_request = StrategyRequestProto(strategy_name=strategy_name)
    self._source_info = source_info
    self._account_request = None

    self._current_ts = \
        get_timestamp() if current_timestamp is None else current_timestamp
    self._start_ts = \
        _get_today_timestamp(self._current_ts) if start_timestamp \
        is None else start_timestamp
    self._daily_reset_disabled = daily_reset_disabled
    self._last_log_ts: Optional[int] = None
    self._last_ticker_ts: Optional[int] = None
    self._pnl_adj_info = None
    self._options_pricing_params: PricingParamsType = {}
    self._reserve_adjust_type = strategy_reporter_proto.UNKNOWN_RESERVE_ADJUST
    self._pnl_balance_details: Optional[List[AccountBalance]] = None

    git_commit_datetime, git_commit_sha = fetch_git_commit_sha_datetime('HEAD')
    self._git_commit_sha = git_commit_sha
    self._git_commit_datetime = \
        int(git_commit_datetime.replace(tzinfo=pytz.UTC).timestamp() * 10**9)

    self._reserve: DefaultDict[AccountKeyType, Dict[str, float]] = \
        defaultdict(dict)
    self._executor_config: Dict[Tuple[AccountKeyType, str],
                                       ExecutorType] = {}
    self._trading_stats: DefaultDict[
        int, Dict[str, TradingStatsCalculator]] = defaultdict(dict)
    self._pnl_balance: Dict[str, PnlFromBalanceCalculator] = {}
    self._pnl_from_fill: DefaultDict[str, List[StrategyStat]] = \
        defaultdict(list)
    self._ticker: Dict[str, float] = {}
    self._ticker_log: Dict[Tuple[str, str], ExchangeTicker] = {}
    self._strategy_info: Dict[str, StrategyInfoProto] = {}
    self._account_info: Dict[AccountRequest, AccountRequestProto] = {}
    self._account_id_request_map: Dict[int, AccountRequestProto] = {}

    self._latency_stats_calculator_1min = \
        LatencyStatsCalculator(self._strategy_request, time_window_sec=60)
    self._latency_stats_calculator_1hr = \
        LatencyStatsCalculator(self._strategy_request, time_window_sec=3600)
    self._logger = logging.getLogger(__name__)
    self._log_num = 0
    self._strat_log_counter = defaultdict(int)
    self._og_log_counter = defaultdict(int)
    self._save_order_event = save_order_event
    self._save_history = save_history
    if strat_stat_log_dir is not None:
      strat_stat_logger = StrategyStatProvider(
          request=StrategyRequestProto(strategy_name=strategy_name),
          log_root=strat_stat_log_dir,
          machine=machine)
      strat_stat_log_dumper = StrategyStatLogDumper(strat_stat_logger)
    else:
      strat_stat_log_dumper = None
    self._strat_stat_log_dumper = strat_stat_log_dumper

  @property
  def pnl_summary(self) -> StrategySummary:
    process = self._gen_process_info_proto()
    strat_info = self._gen_strategy_info_proto()
    calculator_info = self._gen_calculator_info_proto()
    source_info = self._gen_source_info_proto()
    stat_per_account_proto = {}
    for account_id, stats_dict in self._trading_stats.items():
      stat_per_symbol_proto = {symbol: stats.trading_stats for symbol, stats in stats_dict.items()}
      account_request = self._account_id_request_map[account_id]
      acct_stat = AccountStrategyStat(account_request=account_request,
                                      stat_per_symbol=stat_per_symbol_proto)
      stat_per_account_proto[account_id] = acct_stat

    all_order_ts_per_quote = defaultdict(list)
    all_fill_ts_per_quote = defaultdict(list)
    strat_stat = {quote: defaultdict(float) for quote in self._pnl_balance.keys()}
    for stats_dict in self._trading_stats.values():
      for _, stats_calculator in stats_dict.items():
        quote = stats_calculator.accounting_currency
        product_stats = stats_calculator.trading_stats
        if product_stats.HasField('last_order_ts'):
          all_order_ts_per_quote[quote].append(product_stats.last_order_ts)
        if product_stats.HasField('last_fill_ts'):
          all_fill_ts_per_quote[quote].append(product_stats.last_fill_ts)
        if quote not in strat_stat:
          strat_stat[quote] = defaultdict(float)
        stat_per_quote = strat_stat[quote]
        for field in  ['turnover_mark', 'turnover_4hr', 'turnover_maker_mark',
                       'turnover_taker_mark', 'turnover_maker_inferred_mark',
                       'turnover_taker_inferred_mark', 'turnover_mark_in_usd',
                       'turnover_maker_mark_in_usd', 'turnover_taker_mark_in_usd',
                       'turnover_maker_inferred_mark_in_usd',
                       'turnover_taker_inferred_mark_in_usd',
                       'passive_trading_pnl_mark', 'passive_trading_pnl_mark_in_usd',
                       'pnl_mark_from_fill', 'pnl_mark_from_fill_in_usd',
                       'otc_pnl_mark', 'otc_pnl_mark_in_usd',
                       'position_in_usd', 'total_abs_position_in_usd',
                       'avg_abs_position_mark_in_usd', 'fee_mark',
                       'delta_in_usd', 'gamma_in_usd', 'theta_in_usd', 'vega_in_usd']:
          if product_stats.HasField(field):
            stat_per_quote[field] += getattr(product_stats, field)
    stat_per_quote_proto = {}
    for quote, stat in strat_stat.items():
      all_order_ts = all_order_ts_per_quote[quote]
      all_fill_ts = all_fill_ts_per_quote[quote]
      last_order_ts = max(all_order_ts) if len(all_order_ts) > 0 else None
      last_fill_ts = max(all_fill_ts) if len(all_fill_ts) > 0 else None
      pnl_balance = self._get_pnl_balance(quote)
      symbol = '%s-%s' % (quote, 'USD')
      ticker = self._get_ticker(symbol)
      trading_pnl_mark = pnl_balance.pnl_mark
      trading_pnl_4hr = pnl_balance.pnl_4hr
      trading_pnl_mark_in_usd = None if ticker is None else trading_pnl_mark * ticker
      trading_pnl_4hr_in_usd = None if ticker is None else trading_pnl_4hr * ticker
      otc_pnl_mark = stat['otc_pnl_mark']
      otc_pnl_mark_in_usd = stat['otc_pnl_mark_in_usd']
      pnl_mark = trading_pnl_mark + otc_pnl_mark
      active_trading_pnl_mark_in_usd = None
      pnl_mark_in_usd = None
      if trading_pnl_mark_in_usd is not None:
         active_trading_pnl_mark_in_usd = \
             trading_pnl_mark_in_usd - stat['passive_trading_pnl_mark_in_usd']
         pnl_mark_in_usd = trading_pnl_mark_in_usd + otc_pnl_mark_in_usd
      stat_per_quote_proto[quote] = StrategyStat(
          last_log_ts=self._last_log_ts,
          last_order_ts=last_order_ts,
          last_fill_ts=last_fill_ts,
          accounting_currency=quote,
          position_in_usd=stat['position_in_usd'],
          total_abs_position_in_usd=stat['total_abs_position_in_usd'],
          avg_abs_position_mark_in_usd=stat['avg_abs_position_mark_in_usd'],
          fee_mark=stat['fee_mark'],
          pnl_mark=pnl_mark,
          pnl_4hr=trading_pnl_4hr,
          pnl_mark_in_usd=pnl_mark_in_usd,
          pnl_4hr_in_usd=trading_pnl_4hr_in_usd,
          pnl_mark_from_balance=pnl_mark,
          pnl_mark_from_balance_in_usd=pnl_mark_in_usd,
          active_trading_pnl_mark=trading_pnl_mark - stat['passive_trading_pnl_mark'],
          active_trading_pnl_mark_in_usd=active_trading_pnl_mark_in_usd,
          passive_trading_pnl_mark=stat['passive_trading_pnl_mark'],
          passive_trading_pnl_mark_in_usd=stat['passive_trading_pnl_mark_in_usd'],
          pnl_mark_from_fill=stat['pnl_mark_from_fill'],
          pnl_mark_from_fill_in_usd=stat['pnl_mark_from_fill_in_usd'],
          trading_pnl_mark=trading_pnl_mark,
          trading_pnl_mark_in_usd=trading_pnl_mark_in_usd,
          otc_pnl_mark=stat['otc_pnl_mark'],
          otc_pnl_mark_in_usd=stat['otc_pnl_mark_in_usd'],
          turnover_mark=stat['turnover_mark'],
          turnover_4hr=stat['turnover_4hr'],
          turnover_maker_mark=stat['turnover_maker_mark'],
          turnover_taker_mark=stat['turnover_taker_mark'],
          turnover_maker_inferred_mark=stat['turnover_maker_inferred_mark'],
          turnover_taker_inferred_mark=stat['turnover_taker_inferred_mark'],
          turnover_maker_mark_in_usd=stat['turnover_maker_mark_in_usd'],
          turnover_taker_mark_in_usd=stat['turnover_taker_mark_in_usd'],
          turnover_maker_inferred_mark_in_usd=stat['turnover_maker_inferred_mark_in_usd'],
          turnover_taker_inferred_mark_in_usd=stat['turnover_taker_inferred_mark_in_usd'],
          turnover_mark_in_usd=stat.get('turnover_mark_in_usd', None),
          delta_in_usd=stat.get('delta_in_usd'),
          gamma_in_usd=stat.get('gamma_in_usd'),
          theta_in_usd=stat.get('theta_in_usd'),
          vega_in_usd=stat.get('vega_in_usd'),
      )
    latency_stats_1min = _gen_latency_stats_proto(
        process,
        strat_info,
        self._latency_stats_calculator_1min.get_latency_stats(expire_timestamp=self._current_ts))
    """
    latency_stats_1hr = _gen_latency_stats_proto(
        process,
        strat_info,
        self._latency_stats_calculator_1hr.get_latency_stats(expire_timestamp=self._current_ts))
    """
    strat_summary_proto = StrategySummary(process=process,
                                          strategy=strat_info,
                                          calculator=calculator_info,
                                          source=source_info,
                                          mark_ts=self._start_ts,
                                          processing_td=self._start_ts,
                                          source_info=self._source_info,
                                          stat_per_quote=stat_per_quote_proto,
                                          stat_per_account=stat_per_account_proto,
                                          latency_stats_1min=latency_stats_1min)
    return strat_summary_proto

  def _gen_process_info_proto(self) -> ProcessInfoProto:
    td_str = to_datetime(self._start_ts).date().strftime('%Y%m%d')
    process = ProcessInfoProto(trading_date=int(td_str),
                               start_timestamp=self._start_ts,
                               end_timestamp=self._current_ts,
                               machine=self._machine)
    return process

  def _gen_strategy_info_proto(self) -> StrategyInfoProto:
    strategy_info_proto = self._strat_request_to_info(self._strategy_request)
    return strategy_info_proto

  def _gen_calculator_info_proto(self) -> CalculatorInfoProto:
    calculator_proto = CalculatorInfoProto(calculator_name=os.path.basename(__file__),
                                           git_commit_sha=self._git_commit_sha,
                                           git_commit_timestamp=self._git_commit_datetime)
    return calculator_proto

  def _gen_source_info_proto(self) -> StrategySummarySourceInfoProto:
    strat_log_counter = self._strat_log_counter.copy()
    strat_log_counter[StrategyLog.LogType.Name(StrategyLog.OG_LOG)] = \
        self._og_log_counter
    log_num_dict = json.dumps(strat_log_counter)
    strategy_log_info = StrategyLogInfo(
        source_info=self._source_info,
        log_num=self._log_num,
        log_num_dict=log_num_dict)
    strat_log_info = StrategySummarySourceInfoProto(
        type=StrategySummarySourceInfoProto.STRATEGY_LOG,
        strategy_log_info=strategy_log_info)
    return strat_log_info

  def _strat_request_to_info(self, request: StrategyRequestProto) -> StrategyInfoProto:
    assert isinstance(request, StrategyRequestProto), request
    strategy_name = request.strategy_name
    info = StrategyInfoProto(strategy_name=strategy_name)
    if self._strategy_info is not None:
      info = self._strategy_info.get(strategy_name, info)
    return info

  def _get_ticker(self, symbol: str) -> Optional[float]:
    ticker = self._ticker.get(symbol, None)
    if ticker is not None and \
       self._current_ts - self._last_ticker_ts > TICKER_EXPIRE_TIME:
      ticker = None
    return ticker

  def _convert_account_request(self, account_request: AccountRequestProto
  ) -> AccountRequestProto:
    assert isinstance(account_request, AccountRequestProto)
    key = AccountRequest(market_type=account_request.market_type,
                         exchange=account_request.exchange,
                         owner=account_request.owner)
    account_request = self._account_info.get(key, None)
    if account_request is None:
      logging.info('Fail to get account info. %s, %s' % (key, self._strategy_request))
    else:
      assert account_request.HasField('account_id')
    return account_request

  def _get_pnl_balance(self, currency: str) -> PnlFromBalanceCalculator:
    if currency not in self._pnl_balance:
      ticker = {}
      if len(self._ticker_log) > 0:
        for _, exchange_ticker in self._ticker_log.items():
          for each_ticker in exchange_ticker.each_ticker:
            ticker[each_ticker.symbol] = each_ticker.last
      calculator = PnlFromBalanceCalculator(
          strategy_request=self._strategy_request,
          currency=currency,
          ts=self._current_ts,
          start_ts=self._start_ts,
          reserve=self._reserve,
          reserve_adjust_type=self._reserve_adjust_type,
          pnl_balance_details=self._pnl_balance_details,
          ticker=ticker,
          save_history=self._save_history)
      if self._strat_stat_log_dumper is not None:
        calculator.register_strat_stat_observer(
            self._strat_stat_log_dumper.on_strategy_stat_log)
      self._pnl_balance[currency] = calculator
    return self._pnl_balance[currency]

  def _get_trading_stats(self,
                         ts: int,
                         account_request: AccountRequestProto,
                         symbol: str) -> TradingStatsCalculator:
    account_id = account_request.account_id
    trading_stats = self._trading_stats[account_id]
    product = _gen_product(
        account_request.market_type, account_request.exchange, symbol, ts)
    symbol = product.symbol
    if symbol not in trading_stats:
      calculator = TradingStatsCalculator(
            self._strategy_request, account_request, symbol, ts,
            start_ts=self._start_ts,
            save_order_event=self._save_order_event,
            save_history=self._save_history,
            options_pricing_params=self._options_pricing_params)
      trading_stats[symbol] = calculator
      # TODO(chensili): remove this later
      calculator._pnl_from_fill.register_strat_stat_observer(
          self._on_pnl_from_fill_changed)
      if self._strat_stat_log_dumper is not None:
        calculator._pnl_from_fill.register_strat_stat_observer(
            self._strat_stat_log_dumper.on_strategy_stat_log)
        calculator._position_stats.register_strat_stat_observer(
            self._strat_stat_log_dumper.on_strategy_stat_log)
    return trading_stats[symbol]

  def _update_pnl_from_fill(self, ts: int, currency: str):
    if ts < self._start_ts:
      return
    total_pnl_from_fill = 0
    for stats_dict in self._trading_stats.values():
      for _, stats_calculator in stats_dict.items():
        accounting_currency = stats_calculator.accounting_currency
        if accounting_currency != currency:
          continue
        total_pnl_from_fill += stats_calculator.pnl_mark_from_fill
    elem = StrategyStat(stat_ts=ts, pnl_mark_from_fill=total_pnl_from_fill)
    self._pnl_from_fill[currency].append(elem)

  def _on_pnl_from_fill_changed(self, ts: int, strat_stat_log: StrategyStatLog):
    if strat_stat_log.strategy_stat.HasField('pnl_mark_from_fill'):
      self._update_pnl_from_fill(ts, strat_stat_log.accounting_currency)

  def _update_account_id_request_map(self, account_request: AccountRequestProto):
    if account_request is not None:
      if account_request.HasField('account_id'):
        account_id = account_request.account_id
        self._account_id_request_map[account_id] = account_request

  def _update_by_account_request(self, ts: int, account_request: AccountRequestProto):
    assert isinstance(account_request, AccountRequestProto)
    assert account_request.HasField('market_type') and \
        account_request.HasField('exchange') and \
        account_request.HasField('owner')
    account_request = self._convert_account_request(account_request)
    self._update_account_id_request_map(account_request)
    self._account_request = account_request

  def _update_by_balance(self, ts: int, balance: AccountBalance):
    assert isinstance(balance, AccountBalance)
    if self._account_request is not None:
      account_id = self._account_request.account_id
      trading_stats = self._trading_stats[account_id]
      account_key = _get_account_key(self._account_request)
      reserve = self._reserve[account_key]
      if self._account_request.market_type != 'Spot':
        return
      for each_balance in balance.each_balance:
        base = each_balance.currency
        if base not in reserve:
          continue
        for quote in self._pnl_balance.keys():
          for stat in trading_stats.values():
            if (stat.base == base and stat.quote == quote):
              position = each_balance.total - reserve[base]
              stat.update_position(ts, ProductPosition(net_position=position))

  def _update_by_position(
      self, ts: int, position: AccountPosition, *, is_snapshot: bool):
    assert isinstance(position, AccountPosition)
    if self._account_request is not None:
      account_id = self._account_request.account_id
      market_type = self._account_request.market_type
      exchange = self._account_request.exchange
      account_stats = self._trading_stats[account_id]
      symbols = set()
      for each_position in position.each_position:
        symbol = each_position.symbol
        symbols.add(symbol)
        net_position = each_position.net_position
        if symbol not in account_stats:
          if _isclose(net_position, 0):
            continue
          '''
          quote = _get_accounting_currency(market_type, exchange, symbol, ts)
          # Multiple strats may run in one account, but for products with
          # different quote
          if quote not in self._pnl_balance:
            continue
          '''
        trading_stats = self._get_trading_stats(ts, self._account_request, symbol)
        trading_stats.update_position(ts, each_position)
      if is_snapshot:
        for symbol, trading_stats in account_stats.items():
          if symbol not in symbols:
            zero_position = ProductPosition(symbol=symbol, net_position=0)
            trading_stats.update_position(ts, zero_position)

  def _update_by_order_event(self, ts: int, order_event: OrderEvent):
    assert isinstance(order_event, OrderEvent)
    if order_event.type in (OrderEvent.UNKNOWN_ORDER_EVENT,
                            OrderEvent.DEBUG_INFO,):
      return
    if self._account_request is not None:
      symbol = order_event.symbol
      trading_stats = self._get_trading_stats(ts, self._account_request, symbol)
      trading_stats.update_by_order_event(ts, order_event)

  def _handle_strat_request(self, ts: int, pb: StrategyLog):
    assert pb.HasField('request')
    assert pb.request.strategy_name == self._strategy_request.strategy_name

  def _handle_reserve(self, ts: int, pb: StrategyLog):
    assert pb.HasField('reserve')
    if pb.HasField('account_request'):
      account_request = pb.account_request
    else:
      account_request = _extract_account_request(pb.reserve)
    if account_request is not None:
      self._update_by_account_request(ts, account_request)
    if self._account_request is not None:
      for currency, pnl_cal in self._pnl_balance.items():
        pnl_cal.update_by_reserve(ts, pb.reserve, self._account_request)
      account_key = _get_account_key(self._account_request)
      reserve = _get_reserve_dict(pb.reserve)
      self._reserve[account_key] = reserve

  def _handle_pnl_balance(self, ts: int, pb: StrategyLog):
    assert pb.HasField('pnl_balance')
    assert pb.pnl_balance.HasField('currency')
    currency = pb.pnl_balance.currency
    stat = self._get_pnl_balance(currency)
    stat.update_pnl_adjust_info(self._pnl_adj_info)
    stat.update_by_pnl_balance(ts, pb)

  def _handle_pnl_balance_details(self, ts: int, pb: StrategyLog):
    self._pnl_balance_details = pb.pnl_balance_details
    for _, pnl_cal in self._pnl_balance.items():
      pnl_cal.update_by_pnl_balance_details(ts, pb.pnl_balance_details)

  def _handle_executor_config(self, ts: int, pb: StrategyLog):
    if pb.type == StrategyLog.AGG_EXEUCTOR_CONFIG:
      executor_config = pb.agg_executor_config
    elif pb.type == StrategyLog.PASS_EXECUTOR_CONFIG:
      executor_config = pb.pass_executor_config
    else:
      raise ValueError('Unknown executor type: %s' % pb.type)
    account_request = None
    if pb.HasField('account_request'):
      account_request = pb.account_request
    if executor_config.HasField('account_request'):
      account_request = executor_config.account_request
    if account_request is not None:
      self._update_by_account_request(ts, account_request)
      acct_key = _get_account_key(account_request)
      symbol = executor_config.symbol
      product_key = (acct_key, symbol)
      orig_executor_config = self._executor_config.get(product_key)
      if orig_executor_config is None or \
         is_executor_config_updated(orig_executor_config, executor_config):
        self._executor_config[product_key] = executor_config
      else:
        return
    if self._account_request is not None:
      market_type = self._account_request.market_type
      exchange = self._account_request.exchange
      symbol = executor_config.symbol
      try:
        product =_gen_product(market_type, exchange, symbol, ts)
        holder = product.product_holder
      except:
        return
      trading_stats = self._get_trading_stats(ts, self._account_request, symbol)
      trading_stats.update_by_executor_config(ts, executor_config)

  def _handle_og_log(self, ts: int, pb: StrategyLog):
    assert pb.HasField('og_log')
    self._og_log_counter[OrderGatewayLog.LogType.Name(pb.og_log.type)] += 1
    og_log = pb.og_log
    if og_log.HasField('account_request'):
      self._update_by_account_request(ts, og_log.account_request)
    if og_log.type == OrderGatewayLog.BALANCE:
      self._update_by_balance(ts, og_log.balance)
    elif og_log.type == OrderGatewayLog.POSITION:
      self._update_by_position(ts, og_log.position, is_snapshot=True)
    elif og_log.type == OrderGatewayLog.ORDER_EVENT:
      self._update_by_order_event(ts, og_log.event)
    else:
      pass

  def _handle_telemetry(self, ts: int, pb: StrategyLog):
    assert pb.HasField('telemetry')
    self._latency_stats_calculator_1min.update_by_strat_log(ts, pb)
    self._latency_stats_calculator_1hr.update_by_strat_log(ts, pb)

  def _handle_best_book_entry(self, ts: int, pb: StrategyLog):
    assert pb.HasField('best_book_entry')
    best_book_entry = pb.best_book_entry
    if pb.HasField('account_request'):
      self._update_by_account_request(ts, pb.account_request)
    if self._account_request is not None:
      symbol = best_book_entry.symbol
      trading_stats = self._get_trading_stats(ts, self._account_request, symbol)
      trading_stats.update_by_best_book_entry(ts, best_book_entry)

  def _handle_strat_reporter_config(self, ts: int, pb: StrategyLog):
    self._reserve_adjust_type = pb.reporter_config.reserve_adjust_type
    for currency, pnl_cal in self._pnl_balance.items():
      pnl_cal.update_by_strat_reporter_config(ts, pb.reporter_config)

  def _handle_ticker(self, ts: int, pb: StrategyLog):
    assert pb.HasField('ticker')
    self._ticker_log[(pb.ticker.market_type, pb.ticker.exchange)] = pb.ticker
    for currency, pnl_cal in self._pnl_balance.items():
      pnl_cal.update_by_ticker(ts, pb.ticker)
    ticker_dict = {
        (pb.ticker.exchange, each_ticker.symbol): each_ticker
        for each_ticker in pb.ticker.each_ticker
        if pb.ticker.HasField('exchange') and each_ticker.HasField('symbol')
    }
    for _, stats_dict in self._trading_stats.items():
      for symbol, stats in stats_dict.items():
        stats.update_by_ticker(ts, pb.ticker)
        ticker_key = (stats.exchange, symbol)
        product_ticker = ticker_dict.get(ticker_key)
        if stats.market_type == 'Options' and stats.exchange == 'Otc':
          last_pos = stats._position_stats._last_pos
          if last_pos is None:
            continue
          options_value = _calculate_otc_options_price(
              stats.product, last_pos, pb.ticker, ts)
          if options_value is not None:
            product_ticker = ProductTicker(symbol=symbol, last=options_value)
        if product_ticker is not None:
          stats._pnl_from_fill.update_by_ticker(ts, product_ticker)
          stats._passive_pnl.update_by_ticker(ts, product_ticker)

  def _handle_balance_update(self, ts: int, pb: StrategyLog):
    assert pb.HasField('balance_update')
    account_request = _extract_account_request(pb.balance_update)
    if account_request is not None:
      self._update_by_account_request(ts, account_request)
    self._update_by_balance(ts, pb.balance_update)

  def _handle_position_update(self, ts, pb: StrategyLog):
    assert pb.HasField('position_update')
    account_request = _extract_account_request(pb.position_update)
    if account_request is not None:
      self._update_by_account_request(ts, account_request)
    self._update_by_position(ts, pb.position_update, is_snapshot=False)

  def _update_ts(self, current_ts: int, log_ts: int=None):
    if current_ts > self._current_ts:
      self._current_ts = current_ts
    today_ts = _get_today_timestamp(current_ts)
    if not self._daily_reset_disabled:
      if today_ts > self._start_ts:
        self._start_ts = today_ts
    if log_ts is not None:
      if self._last_log_ts is None or log_ts > self._last_log_ts:
        # log ts may be disordered for several reasons:
        # 1. due to network delay when pushed to kafak
        # 2. multi process of same strategy was launched by mistake, leading to
        #    disordered ts in local log file
        self._last_log_ts = log_ts

  def _clean_up_trading_stats(self, ts: int):
    for stats_dict in self._trading_stats.values():
      clean_up_symbol = []
      for symbol, stats in stats_dict.items():
        if not stats.has_stats(ts):
          clean_up_symbol.append(symbol)
      for symbol in clean_up_symbol:
        if symbol in stats_dict:
          del stats_dict[symbol]

  def handle(self, ts: int, pb: StrategyLog):
    self._log_num += 1
    self._strat_log_counter[StrategyLog.LogType.Name(pb.type)] += 1
    ts = int(ts)
    try:
      self._update_ts(ts, log_ts=ts)
      # ignore the log 2 hrs before start_ts
      if ts < self._start_ts - 2 * 60 * 60 * 10**9:
        return
      if pb.type == StrategyLog.STRAT_REQUEST:
        self._handle_strat_request(ts, pb)
      elif pb.type == StrategyLog.RESERVE:
        self._handle_reserve(ts, pb)
      elif pb.type == StrategyLog.PNL_BALANCE:
        self._handle_pnl_balance(ts, pb)
      elif pb.type == StrategyLog.PNL_BALANCE_DETAILS:
        self._handle_pnl_balance_details(ts, pb)
      elif pb.type in (StrategyLog.AGG_EXEUCTOR_CONFIG,
                       StrategyLog.PASS_EXECUTOR_CONFIG):
        self._handle_executor_config(ts, pb)
      elif pb.type == StrategyLog.OG_LOG:
        self._handle_og_log(ts, pb)
      elif pb.type == StrategyLog.TELEMETRY:
        self._handle_telemetry(ts, pb)
      elif pb.type == StrategyLog.BEST_BOOK:
        self._handle_best_book_entry(ts, pb)
      elif pb.type == StrategyLog.STRAT_REPORTER_CONFIG:
        self._handle_strat_reporter_config(ts, pb)
      elif pb.type == StrategyLog.TICKER:
        self._handle_ticker(ts, pb)
      elif pb.type == StrategyLog.BALANCE_UPDATE:
        self._handle_balance_update(ts, pb)
      elif pb.type == StrategyLog.POSITION_UPDATE:
        self._handle_position_update(ts, pb)
    except Exception as e:
      self._logger.info(
          'Fail to handle %s, %s' % (ts, self._strategy_request.strategy_name))
      raise e
    if self._strat_stat_log_dumper is not None:
      self._strat_stat_log_dumper.write()


  def get_result(self, ts=None) -> StrategySummary:
    ts = ts if ts is not None else self._current_ts
    self._update_ts(ts)
    for currency, pnl_cal in self._pnl_balance.items():
      pnl_cal.update_timestamp(self._current_ts, start_ts=self._start_ts)
    for trading_stats_dict in self._trading_stats.values():
      for _, stats in trading_stats_dict.items():
        stats.update_timestamp(self._current_ts, start_ts=self._start_ts)
    self._clean_up_trading_stats(ts)
    return self.pnl_summary

  def get_history(self):
    strat_history = self._get_strat_stats_detail()
    symbol_history = self._get_symbol_stats_detail()
    each_detail = strat_history + symbol_history
    strategy_detail_summary = StrategyDetailSummary(
        process=self._gen_process_info_proto(),
        strategy=self._gen_strategy_info_proto(),
        calculator=self._gen_calculator_info_proto(),
        source_info=self._source_info,
        each_detail=each_detail)
    return strategy_detail_summary

  def get_stat_log(self):
    stat_logs = []
    for account_id, stats_dict in self._trading_stats.items():
      for symbol, stats in stats_dict.items():
        stat_logs += stats.get_stat_log()
    for accounting_currency, pnl_cal in self._pnl_balance.items():
      stat_log_common = StrategyStatLog(
          type=StrategyStatLog.STRATEGY,
          strategy=StrategyInfoProto(strategy_name=self._strategy_request.strategy_name),
          accounting_currency=accounting_currency)
      pnl_balance = _multi_window_pnl_to_strategy_stat(
          pnl_cal.pnl_balance, 'from_balance')
      strat_stats = pnl_balance + self._pnl_from_fill.get(accounting_currency, [])
      for each_stat in strat_stats:
        stat_log = StrategyStatLog(strategy_stat=each_stat)
        stat_log.MergeFrom(stat_log_common)
        stat_logs.append(stat_log)
    stat_logs = sorted(stat_logs, key=lambda x: x.strategy_stat.stat_ts)
    return stat_logs

  def get_log_history(self):
    log_history = {}
    for account_id, stats_dict in self._trading_stats.items():
      for symbol, stats in stats_dict.items():
        acct_req = AccountRequest(market_type=stats._account_request.market_type,
                                  exchange=stats._account_request.exchange,
                                  owner=stats._account_request.owner)
        key = (acct_req, symbol)
        assert key not in log_history
        log_history[key] = stats.get_log_history()
    return log_history

  def get_order_plot_info(self, last=False):
    symbol_plot_msg_list = []
    for account_id, stats_dict in self._trading_stats.items():
      for symbol, stats in stats_dict.items():
        symbol_plot_msg_list.extend(stats.get_order_plot_info(last))
    strategy_order_plot_msg = StrategyOrderPlotMsg(
        strategy_request=self._strategy_request,
        process_info=self._gen_process_info_proto(),
        symbol_plot_msg_list=symbol_plot_msg_list)
    return strategy_order_plot_msg

  def _get_strat_stats_detail(self):
    strat_history = []
    for quote, pnl_cal in self._pnl_balance.items():
      pnl_from_balance = _multi_window_pnl_to_strategy_stat(
          pnl_cal.pnl_balance, 'from_balance')
      pnl_from_fill = self._pnl_from_fill.get(quote)
      strategy_detail = StrategyDetailEntry(type=StrategyDetailEntry.STRATEGY,
                                            accounting_currency=quote,
                                            pnl=pnl_from_balance,
                                            pnl_from_fill=pnl_from_fill)
      strat_history.append(strategy_detail)
    return strat_history

  def _get_symbol_stats_detail(self):
    symbol_history = []
    for account_id, stats_dict in self._trading_stats.items():
      for symbol, stats in stats_dict.items():
        history = stats.get_stats_history()
        symbol_detail = StrategyDetailEntry(
            type=StrategyDetailEntry.SYMBOL,
            account_id=account_id,
            symbol=symbol,
            accounting_currency=stats.accounting_currency,
            pnl=history['pnl'],
            position=history['position'],
            position_from_fill=history['position_from_fill'])
        symbol_history.append(symbol_detail)
    return symbol_history

  def update_ticker(self,
                    ticker: Dict[str, float],
                    ts: Optional[int]=None):
    ts = ts if ts is not None else self._current_ts
    if self._last_ticker_ts is None or ts > self._last_ticker_ts:
      self._last_ticker_ts = ts
      self._ticker = ticker
      for stats_dict in self._trading_stats.values():
        for _, stats in stats_dict.items():
          stats.update_ticker(self._last_ticker_ts, ticker)

  def update_kline(self, kline: List[DBOutputKline]):
    for stats_dict in self._trading_stats.values():
      for _, stats in stats_dict.items():
        stats.update_kline(kline)

  def update_strategy_universe(self, strategy_info: List[StrategyInfoProto]):
    self._strategy_info = {
        each_info.strategy_name: each_info for each_info in strategy_info}

  def update_strategy_info_map(self,
                               strategy_info_map: Dict[str, StrategyInfoProto]):
    self._strategy_info = strategy_info_map

  def update_account_universe(self, account_info: List[AccountInfoProto]):
    info_map = {}
    for each_info in account_info:
      assert each_info.HasField('account_id')
      key = AccountRequest(market_type=each_info.market_type,
                           exchange=each_info.exchange,
                           owner=each_info.owner)
      each_info = AccountRequestProto(account_id=each_info.account_id,
                                      market_type=each_info.market_type,
                                      exchange=each_info.exchange,
                                      owner=each_info.owner)
      info_map[key] = each_info
    self._account_info = info_map

  def update_account_info_map(
      self, info_map: Dict[AccountRequest, AccountRequestProto]):
    self._account_info = info_map

  def update_exchange_api_universe(
      self, exchange_api_info: List[ExchangeApiInfoProto]):
    self._latency_stats_calculator_1min.update_exchange_api_universe(exchange_api_info)
    self._latency_stats_calculator_1hr.update_exchange_api_universe(exchange_api_info)

  def update_pnl_adjust_info(self, pnl_adjust_info):
    self._pnl_adj_info = pnl_adjust_info

  def update_options_pricing_params(self, options_pricing_params: PricingParamsType):
    self._options_pricing_params = options_pricing_params
    for stats_dict in self._trading_stats.values():
      for _, stats in stats_dict.items():
        if stats.market_type == 'Options':
          stats.update_options_pricing_params(options_pricing_params)
